转自:http://xumingming.sinaapp.com/756/twitter-storm-drpc/
-------------------------
Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU intensive的计算。DRPC的storm topology以函数的参数流作为输入,而把这些函数调用的返回值作为topology的输出流。
DRPC其实不能算是storm本身的一个特性, 它是通过组合storm的原语spout,bolt, topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的, 但是DRPC实在是太有用了,所以我们我们把它和storm捆绑在一起。
概览
Distributed RPC是由一个”DPRC Server”协调的(storm自带了一个实现)。DRPC服务器协调1) 接收一个RPC请求。2) 发送请求到storm topology 3) 从storm topology接收结果。4) 把结果发回给等待的客户端。从客户端的角度来看一个DRPC调用跟一个普通的RPC调用没有任何区别。比如下面是客户端如何调用RPC: reach方法的,方法的参数是: http://twitter.com。
- DRPCClientclient=newDRPCClient("drpc-host",3772);
-
Stringresult=client.execute("reach",
-
"http://twitter.com");
DRPCClient client = newDRPCClient("drpc-host",3772);
String result = client.execute("reach",
"http://twitter.com");
DRPC的工作流大致是这样的:
客户端给DRPC服务器发送要执行的方法的名字,以及这个方法的参数。实现了这个函数的topology使用DRPCSpout
从DRPC服务器接收函数调用流。每个函数调用被DRPC服务器标记了一个唯一的id。
这个topology然后计算结果,在topology的最后一个叫做ReturnResults
的bolt会连接到DRPC服务器,并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上,唤醒这个客户端并且把结果发送给它。
LinearDRPCTopologyBuilder
Storm自带了一个称作LinearDRPCTopologyBuilder的topology
builder, 它把实现DRPC的几乎所有步骤都自动化了。这些步骤包括:
-
设置spout
-
把结果返回给DRPC服务器
-
给bolt提供有限聚合几组tuples的能力
让我们看一个简单的例子。下面是一个把输入参数后面添加一个”!”的DRPC topology的实现:
- publicstaticclassExclaimBoltimplementsIBasicBolt{
- publicvoidprepare(Mapconf,TopologyContextcontext){
- }
- publicvoidexecute(Tupletuple,BasicOutputCollectorcollector){
-
Stringinput=tuple.getString(1);
-
collector.emit(newValues(tuple.getValue(0),input+"!"));
- }
- publicvoidcleanup(){
- }
- publicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){
-
declarer.declare(newFields("id","result"));
- }
- }
-
publicstaticvoidmain(String[]args)throwsException{
- LinearDRPCTopologyBuilderbuilder
-
=newLinearDRPCTopologyBuilder("exclamation");
-
builder.addBolt(newExclaimBolt(),3);
-
- }
publicstatic class ExclaimBolt implementsIBasicBolt {
publicvoid prepare(Map conf, TopologyContext context) {
}
publicvoid execute(Tuple tuple, BasicOutputCollector collector) {
String input = tuple.getString(1);
collector.emit(newValues(tuple.getValue(0), input + "!"));
}
publicvoid cleanup() {
}
publicvoid declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(newFields("id","result"));
}
}
publicstatic void main(String[] args) throwsException {
LinearDRPCTopologyBuilder builder
=newLinearDRPCTopologyBuilder("exclamation");
builder.addBolt(newExclaimBolt(), 3);
// ...
}
可以看出来, 我们需要做的事情非常的少。创建LinearDRPCTopologyBuilder
的时候,你需要告诉它你要实现的DRPC函数的名字。一个DRPC服务器可以协调很多函数,函数与函数之间靠函数名字来区分。你声明的第一个bolt会接收两维tuple,tuple的第一个field是request-id,第二个field是这个请求的参数。LinearDRPCTopologyBuilder
同时要求我们topology的最后一个bolt发射一个二维tuple:
第一个field是request-id, 第二个field是这个函数的结果。最后所有中间tuple的第一个field必须是request-id。
在这里例子里面ExclaimBolt
简单地在输入tuple的第二个field后面再添加一个”!”,其余的事情都由LinearDRPCTopologyBuilder
帮我们搞定:链接到DRPC服务器,并且把结果发回。
本地模式DRPC
DRPC可以以本地模式运行。下面就是以本地模式运行上面例子的代码:
- LocalDRPCdrpc=newLocalDRPC();
- LocalClustercluster=newLocalCluster();
- cluster.submitTopology(
-
"drpc-demo",
- conf,
- builder.createLocalTopology(drpc)
- );
-
System.out.println("Resultsfor'hello':"
-
+drpc.execute("exclamation","hello"));
- cluster.shutdown();
- drpc.shutdown();
LocalDRPC drpc = newLocalDRPC();
LocalCluster cluster = newLocalCluster();
cluster.submitTopology(
"drpc-demo",
conf,
builder.createLocalTopology(drpc)
);
System.out.println("Results for 'hello':"
+ drpc.execute("exclamation","hello"));
cluster.shutdown();
drpc.shutdown();
首先你创建一个LocalDRPC
对象。
这个对象在进程内模拟一个DRPC服务器,跟LocalClusterLinearTopologyBuilder
有单独的方法来创建本地的topology和远程的topology。在本地模式里面LocalDRPC
对象不和任何端口绑定,所以我们的topology对象需要知道和谁交互。这就是为什么createLocalTopology
方法接受一个LocalDRPC
对象作为输入的原因。
把topology启动了之后,你就可以通过调用LocalDRPC
对象的execute
来调用RPC方法了。
远程模式DRPC
在一个真是集群上面DRPC也是非常简单的,有三个步骤:
-
启动DRPC服务器
-
配置DRPC服务器的地址
-
提交DRPC topology到storm集群里面去。
我们可以通过下面的storm
脚本命令来启动DRPC服务器:
接着, 你需要让你的storm集群知道你的DRPC服务器在哪里。DRPCSpout
需要这个地址从而可以从DRPC服务器来接收函数调用。这个可以配置在storm.yaml
或者通过代码的方式配置在topology里面。通过storm.yaml
配置是这样的:
- drpc.servers:
-
-"drpc1.foo.com"
-
-"drpc2.foo.com"
drpc.servers:
- "drpc1.foo.com"
- "drpc2.foo.com"
最后,你通过StormSubmitter
对象来提交DRPC
topology — 跟你提交其它topology没有区别。如果要以远程的方式运行上面的例子,用下面的代码:
- StormSubmitter.submitTopology(
-
"exclamation-drpc",
- conf,
- builder.createRemoteTopology()
- );
StormSubmitter.submitTopology(
"exclamation-drpc",
conf,
builder.createRemoteTopology()
);
我们用createRemoteTopology
方法来创建运行在真实集群上的DRPC
topology。
一个更复杂的例子
上面的DRPC例子只是为了介绍DRPC概念的一个简单的例子。下面让我们看一个复杂的、确实需要storm的并行计算能力的例子, 这个例子计算twitter上面一个url的reach值。
首先介绍一下什么是reach值,要计算一个URL的reach值,我们需要:
-
获取所有微薄里面包含这个URL的人
-
获取这些人的粉丝
-
把这些粉丝去重
-
获取这些去重之后的粉丝个数 — 这就是reach
一个简单的reach计算可能会有成千上万个数据库调用,并且可能设计到百万数量级的微薄用户。这个确实可以说是CPU intensive的计算了。你会看到的是,在storm上面来实现这个是非常非常的简单。在单台机器上面, 一个reach计算可能需要花费几分钟。而在一个storm集群里面,即时是最男的URL, 也只需要几秒。
一个reach topolgoy的例子可以在这里找到(storm-starter)。reach
topology是这样定义的:
- LinearDRPCTopologyBuilderbuilder
-
=newLinearDRPCTopologyBuilder("reach");
-
builder.addBolt(newGetTweeters(),3);
-
builder.addBolt(newGetFollowers(),12)
- .shuffleGrouping();
-
builder.addBolt(newPartialUniquer(),6)
-
.fieldsGrouping(newFields("id","follower"));
-
builder.addBolt(newCountAggregator(),2)
-
.fieldsGrouping(newFields("id"));
LinearDRPCTopologyBuilder builder
=newLinearDRPCTopologyBuilder("reach");
builder.addBolt(newGetTweeters(), 3);
builder.addBolt(newGetFollowers(), 12)
.shuffleGrouping();
builder.addBolt(newPartialUniquer(), 6)
.fieldsGrouping(newFields("id","follower"));
builder.addBolt(newCountAggregator(), 2)
.fieldsGrouping(newFields("id"));
这个topology分四步执行:
-
GetTweeters
获取所发微薄里面包含制定URL的所有用户。它接收输入流:[id,
url]
, 它输出:[id,
tweeter]
. 没一个URL tuple会对应到很多tweeter
tuple。
-
GetFollowers
获取这些tweeter的粉丝。它接收输入流:[id,
tweeter]
, 它输出:[id,
follower]
-
PartialUniquer
通过粉丝的id来group粉丝。这使得相同的分析会被引导到统一个task。因此不同的task接收到的粉丝是不同的
— 从而起到去重的作用。它的输出流:[id,
count]
即输出这个task上统计的粉丝个数。
-
最后,
CountAggregator
接收到所有的局部数量,
把它们加起来就算出了我们要的reach值。
我们来看一下PartialUniquer
的实现:
- publicstaticclassPartialUniquer
- implementsIRichBolt,FinishedCallback{
- OutputCollector_collector;
- Map<Object,Set<String>>_sets
- =newHashMap<Object,Set<String>>();
- publicvoidprepare(Mapconf,
- TopologyContextcontext,
- OutputCollectorcollector){
- _collector=collector;
- }
- publicvoidexecute(Tupletuple){
-
Objectid=tuple.getValue(0);
- Set<String>curr=_sets.get(id);
-
if(curr==null){
- curr=newHashSet<String>();
- _sets.put(id,curr);
- }
-
curr.add(tuple.getString(1));
- _collector.ack(tuple);
- }
- publicvoidcleanup(){
- }
- publicvoidfinishedId(Objectid){
- Set<String>curr=_sets.remove(id);
- intcount;
-
if(curr!=null){
- count=curr.size();
-
}else{
-
count=0;
- }
- _collector.emit(newValues(id,count));
- }
- publicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){
-
declarer.declare(newFields("id","partial-count"));
- }
- }
publicstatic class PartialUniquer
implementsIRichBolt, FinishedCallback {
OutputCollector _collector;
Map<Object, Set<String>> _sets
=newHashMap<Object, Set<String>>();
publicvoid prepare(Map conf,
TopologyContext context,
OutputCollector collector) {
_collector = collector;
}
publicvoid execute(Tuple tuple) {
Object id = tuple.getValue(0);
Set<String> curr = _sets.get(id);
if(curr==null) {
curr = newHashSet<String>();
_sets.put(id, curr);
}
curr.add(tuple.getString(1));
_collector.ack(tuple);
}
publicvoid cleanup() {
}
publicvoid finishedId(Object id) {
Set<String> curr = _sets.remove(id);
intcount;
if(curr!=null) {
count = curr.size();
}else{
count = 0;
}
_collector.emit(newValues(id, count));
}
publicvoid declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(newFields("id","partial-count"));
}
}
当PartialUniquer
在execute
方法里面接收到一个粉丝tuple
的时候,
它把这个tuple添加到当前request-id对应的Set
里面去。
PartialUniquer
同时也实现了FinishedCallback
接口,
实现这个接口是告诉LinearDRPCTopologyBuilder
它想在接收到某个request-id的所有tuple之后得到通知,回调函数则是,code>finishedId方法。在这个回调函数里面PartialUniquer
发射当前这个request-id在这个task上的粉丝数量。
在这个简单接口的背后,我们是使用CoordinatedBolt
来检测什么时候一个bolt接收到某个request的所有的tuple的。CoordinatedBolt
是利用direct
stream来实现这种协调的。
这个topology的其余部分就非常的明了了。我们可以看到的是reach计算的每个步骤都是并行计算出来的,而且实现这个DRPC的topology是那么的简单。
非线性DRPC Topology
LinearDRPCTopologyBuilder
只能搞定"线性"的DRPC
topology。所谓的线性就是说你的计算过程是一步接着一步, 串联。我们不难想象还有其它的可能 -- 并联(回想一下初中物理里面学的并联电路吧), 现在你如果想解决这种这种并联的case的话, 那么你需要自己去使用CoordinatedBolt
来处理所有的事情了。如果真的有这种use
case的话, 在mailing list上大家讨论一下吧。
LinearDRPCTopologyBuilder的工作原理
-
DRPCSpout发射tuple:
[args,
return-info]
。return-info
包含DRPC服务器的主机地址,端口以及当前请求的request-id
-
DRPC Topology包含以下元素:
-
DRPCSpout
-
PrepareRequest(生成request-id, return info以及args)
-
CoordinatedBolt
-
JoinResult -- 组合结果和return info
-
ReturnResult -- 连接到DRPC服务器并且返回结果
-
LinearDRPCTopologyBuilder是利用storm的原语来构建高层抽象的很好的例子。
高级特性
分享到:
相关推荐
Storm-drpc节点适用于Node.js的Apache Storm DRPC客户端受启发,但不同之处在于可以选择将其设置为保持活动状态,它不需要在每个execute()调用中都创建连接,并且可以喜欢的传统方式或promise方式使用它。...
storm之drpc操作demo示例
这是storm中drpc应用的一个例子。
storm DRPC简单例程,服务器端是运行在集群环境中的,客户端去调用DRPC服务
DRPC 与传输无关,支持中间件,并围绕接口设计。 战斗测试。 已在数万台服务器的生产中使用多年。 外包装 go.bryk.io/pkg/net/drpc 简化的 TLS 设置(用于客户端和服务器) 服务器中间件,包括日志记录、基于...
第6章~第8章详细而系统地讲解了几个高级特性:事务、DRPC和Trident;第9章以实例的方式讲解了Storm在实际业务场景中的应用;第10章总结了几个在大数据场景应用过程中遇到的经典问题,以及详细的排查过程。
01.Storm基础知识02.Storm集群安装-1-new .avi.baiduyun.p05.Storm配置文件配置项讲解07.Storm基本API介绍08.Storm Topology的并发度09.Strom消息机制原理讲解10.Storm DRPC实战讲解
内容概要: • Storm 记录级容错原理 • Storm 配置详解 • Storm 批处理 • Storm TOPN • Storm 流程聚合 • Storm DRPC • Storm executor、worker、task之间的关系和调优 • Storm异常解决
• BasicBolt • Storm 批处理 • Storm TOPN • Storm 流程聚合 • Storm DRPC • Storm executor、worker、task之间的关系和调优 • Storm异常解决
Storm入门教程 之Storm原理和概念详解,出自Storm流计算从入门到精通之技术篇,Storm入门视频教程用到技术:Storm集群、Zookeeper集群等,涉及项目:网站PV、UV案例实战、其他案例; Storm视频教程亮点: 1、Storm...
目前包括几个服务: storm-mesos - Nimbus(Mesos 调度程序) storm-ui - Web 界面(默认端口为8080 ) storm-drpc - DRPC 守护进程storm-log - 用于在 Web UI 中显示日志的服务(从计算节点获取日志) 请注意,不...
【Storm篇】--Storm中的同步服务DRPC 【Storm篇】--Storm从初始到分布式搭建 【Storm篇】--Storm 容错机制 【Storm篇】--Storm并发机制 【Storm篇】--Storm分组策略 【Storm篇】--Storm基础概念
风暴Debian包装 用于分布式实时计算系统的... 提供以下服务的软件包: storm-drpc storm-logviewer storm-nimbus storm-supervisor storm-ui 还有storm-common软件包,它是服务软件包的依赖项。 还有一个storm软件包,
3、注重实践,对较抽象难懂的技术点如Grouping策略、并发度及线程安全、批处理事务、DRPC、Storm Trident均结合企业场景开发案例进行讲解,让学员觉得简单易懂; 4、每个技术均采用最新稳定版本,学完后会员可以从...
DRPC:简单的Discord RPC程序
linux实验环境,storm搭建完毕后的开发。eclipse开发环境,大数据界hello world——wordcount详解,bolt、分组机制、storm DRPC详解
dRPC 一些描述。要求移液器> = 0.1.4安装pip install dRPC特征一些功能。文献资料例子一些描述。 # Some code 新分支测试
分布式RPC(distributedRPC,DRPC)用于对Storm上大量的函数调用进行并行计算过程。对于每一次函数调用,Storm集群上运行的拓扑接收调用函数的参数信息作为输入流,并将计算结果作为输出流发射出去。DRPC本身算不上...