转 http://www.cnblogs.com/panfeng412/archive/2012/07/02/storm-common-patterns-of-distributed-rpc.html
本文翻译自:https://github.com/nathanmarz/storm/wiki/Distributed-RPC,作为学习Storm DRPC的资料,转载必须以超链接形式标明文章原始出处及本文翻译链接。
分布式RPC(distributed RPC,DRPC)用于对Storm上大量的函数调用进行并行计算过程。对于每一次函数调用,Storm集群上运行的拓扑接收调用函数的参数信息作为输入流,并将计算结果作为输出流发射出去。
DRPC本身算不上Storm的特性,它是通过Storm的基本元素:streams,spouts,bolts,topologies而衍生的一个模式。DRPC可以单独作为一个独立于Storm的库发布,但由于其重要性还是和Storm捆绑在了一起。
总体概述
DRPC通过DRPC Server来实现,DRPC Server的整体工作过程如下:
- 接收到一个RPC调用请求;
- 发送请求到Storm上的拓扑;
- 从Storm上接收计算结果;
- 将计算结果返回给客户端。
以上过程,在client客户端看来,一个DRPC调用看起来和一般的RPC调用没什么区别。下面代码是client通过DRPC调用“reach”函数,参数为“http://twitter.com”:
DRPCClient client = new DRPCClient("drpc-host", 3772); String result = client.execute("reach", "http://twitter.com");
DRPC内部工作流程如下:
- Client向DRPC Server发送被调用执行的DRPC函数名称及参数。
- Storm上的topology通过DRPCSpout实现这一函数,从DPRC Server接收到函数调用流;
- DRPC Server会为每次函数调用生成唯一的id;
- Storm上运行的topology开始计算结果,最后通过一个ReturnResults的Bolt连接到DRPC Server,发送指定id的计算结果;
- DRPC Server通过使用之前为每个函数调用生成的id,将结果关联到对应的发起调用的client,将计算结果返回给client。
LinearDRPCTopologyBuilder
Storm提供了一个topology builder——LinearDRPCTopologyBuilder,它可以自动完成几乎所有的DRPC步骤。包括:
- 构建spout;
- 向DRPC Server返回结果;
- 为Bolt提供函数用于对tuples进行聚集。
下面是一个简单的例子,这个DRPC拓扑只是简单的在输入参数后追加“!”后返回:
public static class ExclaimBolt extends BaseBasicBolt { public void execute(Tuple tuple, BasicOutputCollector collector) { String input = tuple.getString(1); collector.emit(new Values(tuple.getValue(0), input + "!")); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "result")); } } public static void main(String[] args) throws Exception { LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation"); builder.addBolt(new ExclaimBolt(), 3); // ... }
由上述例子可见,我们只需很少的工作即可完成拓扑。当创建LinearDRPCTopologyBuilder
的时候,需要指定拓扑中
DRPC
函数的名称
“
exclamation”
。一个
DRPC Server
可以协调多个函数,每个函数有不同的函数名称。拓扑中的第一个
bolt
的输入是
两
个字段:第一个是请求的
id
号;第二个是请求的参数。
LinearDRPCTopologyBuilder
同时需要最后一个
bolt
发射一个包含两个字段的输出流:第一个字段是请求
id
;第二个字段是计算结果。因此,所有的中间
tuples
必须包含请求
id
作为第一个字段。
例子中,
ExclaimBolt
在输入
tuple
的第二个字段后面追加
“!”
,
LinearDRPCTopologyBuilder
负责处理其余的协调工作:与
DRPC Server
建立连接,发送结果给
DRPC Server
。
本地模式DRPC
DRPC可以以本地模式运行,下面的代码是如何在本地模式运行上面的例子:
LocalDRPC drpc = new LocalDRPC(); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc)); System.out.println("Results for 'hello':" + drpc.execute("exclamation", "hello")); cluster.shutdown(); drpc.shutdown();
首先创建一个LocalDRPC
对象,该对象在本地模拟一个
DRPC Server
,正如
LocalCluster
在本地模拟一个
Storm
集群一样。然后创建一个
LocalCluster
对象在本地模式下运行拓扑。
LinearDRPCTopologyBuilder
含有单独的方法用于创建本地拓扑和远程拓扑。
本地模式下,
LocalDRPC
并不绑定任何端口,因此
Storm
的拓扑需要了解要通讯的对象
——
这就是为什么
createLocalTopology
方法需要以
LocalDRPC
对象作为输入。
加载完拓扑之后,通过对
LocalDRPC
调用
execute
方法,就可以执行
DRPC
函数调用了。
远程模式DRPC
在实际的Storm集群上运行DRPC也一样很简单。只需完成以下步骤:
- 启动DRPC Server(s);
- 配置DRPC Server(s)地址;
- 向Storm集群提交DRPC拓扑。
首先,通过storm脚本启动DRPC Server:
bin/storm drpc
然后,在Storm集群中配置DRPC Server地址,这就是DRPCSpout
读取函数调用请求的地方。这一步的配置可以通过
storm.yaml
文件或者拓扑的配置来完成。通过
storm.yaml
文件的配置方式如下:
drpc.servers:
- "drpc1.foo.com"
- "drpc2.foo.com"
最后,通过StormSubmitter
启动
DRPC
拓扑。为了以远程模式运行
上面的例子,代码如下:
StormSubmitter.submitTopology("exclamation-drpc", conf, builder.createRemoteTopology());
createRemoteTopology
被用于为
Storm
集群创建合适的拓扑。
一个复杂的例子
上面的exclamation只是一个简单的DRPC例子。下面通过一个复杂的例子介绍如何在Storm集群内进行DRPC——计算Twitter上每个URL的到达度(reach),也就是每个URL暴露给的不同人的个数。
为了完成这一计算,需要完成以下步骤:
- 获取所有点选了(tweet)该URL的人;
- 获取步骤1中所有人的关注者(followers,粉丝);
- 对所有关注者followers进行去重;
- 对步骤3中的关注者人数进行求和。
一个简单的URL到达度计算可能涉及成千上万次数据库调用以及数以百万的followers记录,计算量非常大。有了Storm,将很容易实现这一计算过程。单机上可能需要运行几分钟才能完成,在Storm集群上,即使是最难计算的URL也只需要几秒钟。
这个例子的代码在storm-starter:点击这里。这里是如何创建拓扑的代码:
LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach"); builder.addBolt(new GetTweeters(), 3); builder.addBolt(new GetFollowers(), 12) .shuffleGrouping(); builder.addBolt(new PartialUniquer(), 6) .fieldsGrouping(new Fields("id", "follower")); builder.addBolt(new CountAggregator(), 2) .fieldsGrouping(new Fields("id"));
拓扑的执行分为以下四步:
- GetTweeters:获取所有tweet了指定URL的用户列表,这个Bolt将输入流[id, url]转换成输出流[id, tweeter],每个url元组被映射为多个tweeter元组。
- GetFollowers:获取步骤1中所有用户列表的followers,这个Bolt将输入流[id, twetter]转换成输出流[id, follower],当某个人同时是多个人的关注者follower,而且这些人都tweet了指定的URL,那么将产生重复的follower元组。
- PartialUniquer:将所有followers按照follower id分组,使得同一个follower在同一个task中被处理。这个Bolt接收follower并进行去重计数。
- CountAggregator:从各个PartialUniquer中接收各部分的计数结果,累加后完成到达度计算。
下面是PartialUniquer
这个
Bolt
的代码实现:
public class PartialUniquer extends BaseBatchBolt { BatchOutputCollector _collector; Object _id; Set<String> _followers = new HashSet<String>(); @Override public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) { _collector = collector; _id = id; } @Override public void execute(Tuple tuple) { _followers.add(tuple.getString(1)); } @Override public void finishBatch() { _collector.emit(new Values(_id, _followers.size())); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "partial-count")); } }
PartialUniquer
通过继承
BaseBatchBolt
实现了
IBatchBolt
接口,
batch bolt
提供了
API
用于将一批
tuples
作为整体来处理。每个请求
id
会创建一个新的
batch bolt
实例,同时
Storm
负责这些实例的清理工作。
当PartialUniquer
接收到一个
follower
元组时执行
execute
方法,将
follower
添加到请求
id
对应的
HashSet
集合中。
Batch bolt同时提供了finishBatch
方法用于当这个
task
已经处理完所有的元组时调用。
PartialUniquer
发射一个包含当前
task
所处理的
follower ids
子集去重后个数的元组。
在内部实现上,
CoordinatedBolt
用于检测指定的
bolt
是否已经收到指定请求
id
的所有
tuples
元组。
CoordinatedBolt
使用
direct streams管理实现这一协作过程。
拓扑的其他部分易于理解。到达度的每一步的计算过程都是并行进行的,通过DRPC实现也是非常容易的。
Non-linear DRPC拓扑
LinearDRPCTopologyBuilder
只能处理
“
线性的
”DRPC
拓扑
——
正如到达度这样可以通过一系列步骤序列来完成的计算。不难想象,
DRPC
调用中包含有更复杂的带有分支和合并
Bolt
的拓扑。目前,必须自己直接使用
CoordinatedBolt
来完成这种非线性拓扑的计算。
LinearDRPCTopologyBuilder工作过程
- DRPCSpout发射[args, return-info],其中return-info包含DRPC Server的主机和端口号,以及DRPC Server为该次请求生成的唯一id号;
- 构造一个Storm拓扑包含以下部分:
- DRPCSpout
- PrepareRequest(生成一个请求id,为return info创建一个流,为args创建一个流)
- CoordinatedBolt wrappers以及direct groupings
- JoinResult(将结果与return info拼接起来)
- ReturnResult(连接到DRPC Server,返回结果)
- LinearDRPCTopologyBuilder是建立在Storm基本元素之上的高层抽象。
高级进阶
- KeyedFairBolt用于组织同一时刻多请求的处理过程;
- 如何直接使用
CoordinatedBolt
。
相关推荐
storm DRPC简单例程,服务器端是运行在集群环境中的,客户端去调用DRPC服务
Storm-drpc节点适用于Node.js的Apache Storm DRPC客户端受启发,但不同之处在于可以选择将其设置为保持活动状态,它不需要在每个execute()调用中都创建连接,并且可以喜欢的传统方式或promise方式使用它。...
这是storm中drpc应用的一个例子。
storm之drpc操作demo示例
01.Storm基础知识02.Storm集群安装-1-new .avi.baiduyun.p05.Storm配置文件配置项讲解07.Storm基本API介绍08.Storm Topology的并发度09.Strom消息机制原理讲解10.Storm DRPC实战讲解
• BasicBolt • Storm 批处理 • Storm TOPN • Storm 流程聚合 • Storm DRPC • Storm executor、worker、task之间的关系和调优 • Storm异常解决
内容概要: • Storm 记录级容错原理 • Storm 配置详解 • Storm 批处理 • Storm TOPN • Storm 流程聚合 • Storm DRPC • Storm executor、worker、task之间的关系和调优 • Storm异常解决
linux实验环境,storm搭建完毕后的开发。eclipse开发环境,大数据界hello world——wordcount详解,bolt、分组机制、storm DRPC详解
2、注重实践,对较抽象难懂的技术点如Grouping策略、并发度及线程安全、批处理事务、DRPC、Storm Trident均结合企业场景开发案例进行讲解,简单易懂; 3、分享积累的经验和技巧,从架构的角度剖析场景和设计实现...
3、注重实践,对较抽象难懂的技术点如Grouping策略、并发度及线程安全、批处理事务、DRPC、Storm Trident均结合企业场景开发案例进行讲解,让学员觉得简单易懂; 4、每个技术均采用最新稳定版本,学完后会员可以从...
第6章~第8章详细而系统地讲解了几个高级特性:事务、DRPC和Trident;第9章以实例的方式讲解了Storm在实际业务场景中的应用;第10章总结了几个在大数据场景应用过程中遇到的经典问题,以及详细的排查过程。
【Storm篇】--Storm中的同步服务DRPC 【Storm篇】--Storm从初始到分布式搭建 【Storm篇】--Storm 容错机制 【Storm篇】--Storm并发机制 【Storm篇】--Storm分组策略 【Storm篇】--Storm基础概念
目前包括几个服务: storm-mesos - Nimbus(Mesos 调度程序) storm-ui - Web 界面(默认端口为8080 ) storm-drpc - DRPC 守护进程storm-log - 用于在 Web UI 中显示日志的服务(从计算节点获取日志) 请注意,不...
风暴Debian包装 用于分布式实时计算系统的... 提供以下服务的软件包: storm-drpc storm-logviewer storm-nimbus storm-supervisor storm-ui 还有storm-common软件包,它是服务软件包的依赖项。 还有一个storm软件包,
分布式RPC(distributedRPC,DRPC)用于对Storm上大量的函数调用进行并行计算过程。对于每一次函数调用,Storm集群上运行的拓扑接收调用函数的参数信息作为输入流,并将计算结果作为输出流发射出去。DRPC本身算不上...
要遵循本教程,我假设您知道Storm和Trident是什么,并且对它们的核心抽象(即Spouts,Bolts,Stream等)有足够的了解。此外,我假设您知道什么是nimbus和supervisor 。 拓扑和DRPC流 拓扑简单地从输入喷嘴中读取...
JEA主要特点如下:1、DRPC,分布式远程过程调用,通过Storm实现,序列化由Kryo支持。2、L2二级缓存的支持,考虑到应用分布式部署的原因,数据只能通过Memcached和Redis进行缓存。3、Hibernate和Mybatis的集成,DB的...