`
dawuafang
  • 浏览: 1156786 次
文章分类
社区版块
存档分类
最新评论

Twitter Storm: DRPC简介 .

 
阅读更多

转自: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。

  1. DRPCClientclient=newDRPCClient("drpc-host",3772);
  2. Stringresult=client.execute("reach",
  3. "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的实现:

  1. publicstaticclassExclaimBoltimplementsIBasicBolt{
  2. publicvoidprepare(Mapconf,TopologyContextcontext){
  3. }
  4. publicvoidexecute(Tupletuple,BasicOutputCollectorcollector){
  5. Stringinput=tuple.getString(1);
  6. collector.emit(newValues(tuple.getValue(0),input+"!"));
  7. }
  8. publicvoidcleanup(){
  9. }
  10. publicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){
  11. declarer.declare(newFields("id","result"));
  12. }
  13. }
  14. publicstaticvoidmain(String[]args)throwsException{
  15. LinearDRPCTopologyBuilderbuilder
  16. =newLinearDRPCTopologyBuilder("exclamation");
  17. builder.addBolt(newExclaimBolt(),3);
  18. //...
  19. }

可以看出来, 我们需要做的事情非常的少。创建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可以以本地模式运行。下面就是以本地模式运行上面例子的代码:

  1. LocalDRPCdrpc=newLocalDRPC();
  2. LocalClustercluster=newLocalCluster();
  3. cluster.submitTopology(
  4. "drpc-demo",
  5. conf,
  6. builder.createLocalTopology(drpc)
  7. );
  8. System.out.println("Resultsfor'hello':"
  9. +drpc.execute("exclamation","hello"));
  10. cluster.shutdown();
  11. drpc.shutdown();

首先你创建一个LocalDRPC对象。 这个对象在进程内模拟一个DRPC服务器,跟LocalClusterLinearTopologyBuilder有单独的方法来创建本地的topology和远程的topology。在本地模式里面LocalDRPC对象不和任何端口绑定,所以我们的topology对象需要知道和谁交互。这就是为什么createLocalTopology方法接受一个LocalDRPC对象作为输入的原因。


把topology启动了之后,你就可以通过调用LocalDRPC对象的execute来调用RPC方法了。

远程模式DRPC

在一个真是集群上面DRPC也是非常简单的,有三个步骤:

  • 启动DRPC服务器
  • 配置DRPC服务器的地址
  • 提交DRPC topology到storm集群里面去。

我们可以通过下面的storm脚本命令来启动DRPC服务器:

  1. bin/stormdrpc

接着, 你需要让你的storm集群知道你的DRPC服务器在哪里。DRPCSpout需要这个地址从而可以从DRPC服务器来接收函数调用。这个可以配置在storm.yaml或者通过代码的方式配置在topology里面。通过storm.yaml配置是这样的:

  1. drpc.servers:
  2. -"drpc1.foo.com"
  3. -"drpc2.foo.com"

最后,你通过StormSubmitter对象来提交DRPC topology — 跟你提交其它topology没有区别。如果要以远程的方式运行上面的例子,用下面的代码:

  1. StormSubmitter.submitTopology(
  2. "exclamation-drpc",
  3. conf,
  4. builder.createRemoteTopology()
  5. );

我们用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是这样定义的:

  1. LinearDRPCTopologyBuilderbuilder
  2. =newLinearDRPCTopologyBuilder("reach");
  3. builder.addBolt(newGetTweeters(),3);
  4. builder.addBolt(newGetFollowers(),12)
  5. .shuffleGrouping();
  6. builder.addBolt(newPartialUniquer(),6)
  7. .fieldsGrouping(newFields("id","follower"));
  8. builder.addBolt(newCountAggregator(),2)
  9. .fieldsGrouping(newFields("id"));

这个topology分四步执行:

  • GetTweeters获取所发微薄里面包含制定URL的所有用户。它接收输入流:[id, url], 它输出:[id, tweeter]. 没一个URL tuple会对应到很多tweetertuple。
  • GetFollowers获取这些tweeter的粉丝。它接收输入流:[id, tweeter], 它输出:[id, follower]
  • PartialUniquer通过粉丝的id来group粉丝。这使得相同的分析会被引导到统一个task。因此不同的task接收到的粉丝是不同的 — 从而起到去重的作用。它的输出流:[id, count]即输出这个task上统计的粉丝个数。
  • 最后,CountAggregator接收到所有的局部数量, 把它们加起来就算出了我们要的reach值。

我们来看一下PartialUniquer的实现:

  1. publicstaticclassPartialUniquer
  2. implementsIRichBolt,FinishedCallback{
  3. OutputCollector_collector;
  4. Map<Object,Set<String>>_sets
  5. =newHashMap<Object,Set<String>>();
  6. publicvoidprepare(Mapconf,
  7. TopologyContextcontext,
  8. OutputCollectorcollector){
  9. _collector=collector;
  10. }
  11. publicvoidexecute(Tupletuple){
  12. Objectid=tuple.getValue(0);
  13. Set<String>curr=_sets.get(id);
  14. if(curr==null){
  15. curr=newHashSet<String>();
  16. _sets.put(id,curr);
  17. }
  18. curr.add(tuple.getString(1));
  19. _collector.ack(tuple);
  20. }
  21. publicvoidcleanup(){
  22. }
  23. publicvoidfinishedId(Objectid){
  24. Set<String>curr=_sets.remove(id);
  25. intcount;
  26. if(curr!=null){
  27. count=curr.size();
  28. }else{
  29. count=0;
  30. }
  31. _collector.emit(newValues(id,count));
  32. }
  33. publicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){
  34. declarer.declare(newFields("id","partial-count"));
  35. }
  36. }

PartialUniquerexecute方法里面接收到一个粉丝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操作demo示例.zip

    总结来说,"storm之drpc操作demo示例"是一个很好的学习资源,它涵盖了Storm DRPC的核心概念和实践操作,对于想要在实时计算项目中运用Storm DRPC功能的开发者来说,极具参考价值。通过实际操作这个示例,你将能够...

    storm-drpc-node:适用于Node.js的Apache Storm DRPC客户端

    Storm-drpc节点适用于Node.js的Apache Storm DRPC客户端受启发,但不同之处在于可以选择将其设置为保持活动状态,它不需要在每个execute()调用中都创建连接,并且可以喜欢的传统方式或promise方式使用它。...

    storm DRPC简单例程

    5. **集群环境**:Storm集群通常由多个节点组成,包括nimbus(主控节点)、supervisors(工作节点)和Zookeeper(协调服务)。nimbus负责调度任务,supervisors执行任务,而Zookeeper确保集群的高可用性。 6. **...

    Storm配置项详解.docx

    20. **drpc.port**:Storm的DRPC服务端口。 21. **supervisor.slots.ports**:supervisor上可用于运行workers的端口列表,每个worker占用一个端口。 22. **supervisor.childopts**:配置supervisor守护进程的JVM...

    Storm的drpc应用

    这是storm中drpc应用的一个例子。

    02、Storm入门到精通storm3-0.pptx

    - **流程聚合**:Storm支持通过Bolt实现复杂的计算,如聚合、分组等操作,对数据流进行处理。 - **Storm DRPC**:DRPC允许用户在Storm拓扑中直接执行远程过程调用,提供实时计算服务。 - **Executor、Worker、Task...

    02、Storm入门到精通storm3-1.pptx

    在深入学习Storm时,还需要关注其他概念,如`批处理(Batch Processing)`、`TOPN`(用于计算Top N个元素)、`DRPC(Distributed Remote Procedure Calls)`以及`Executor、Worker、Task`之间的关系和调优。...

    Storm深入学习.pdf

    - **Storm TOPN**:Storm 提供了 TopN 组件,用于计算数据流中的前 N 个元素,如最常见的单词、最高点击率等。 - **流程聚合**:Bolt 可以聚合 Tuple 数据,执行各种计算,如求和、平均值、最大值等。 - **DRPC...

    drpc:drpc 是 gRPC 的轻量级替代品

    DRPC 与传输无关,支持中间件,并围绕接口设计。 战斗测试。 已在数万台服务器的生产中使用多年。 外包装 go.bryk.io/pkg/net/drpc 简化的 TLS 设置(用于客户端和服务器) 服务器中间件,包括日志记录、基于...

    Storm配置详解

    17. drpc.servers和drpc.ports:DRPC(分布式远程过程调用)服务器列表和服务端口。 18. supervisor.slots.ports:Supervisor上能运行workers的端口列表。每个worker占用一个端口,并且每个端口只运行一个worker。...

    Storm实战:构建大数据实时计算

    第6章~第8章详细而系统地讲解了几个高级特性:事务、DRPC和Trident;第9章以实例的方式讲解了Storm在实际业务场景中的应用;第10章总结了几个在大数据场景应用过程中遇到的经典问题,以及详细的排查过程。

    storm集群部署和配置过程详解

    根据具体需求,可能还需要配置其他的组件,如drpc(分布式RPC)或logviewer。 在实际部署中,还需要考虑网络拓扑,确保nimbus和worker之间的通信畅通。同时,为了保证系统的稳定性和性能,需要监控资源使用情况,如...

    大数据平台Storm入门到精通

    01.Storm基础知识02.Storm集群安装-1-new .avi.baiduyun.p05.Storm配置文件配置项讲解07.Storm基本API介绍08.Storm Topology的并发度09.Strom消息机制原理讲解10.Storm DRPC实战讲解

    storm深入学习.pdf

    深入学习Storm,我们需要理解其核心概念,包括数据模型(topology)、容错机制(ack和fail)、批处理、TOPN算法、流程聚合、DRPC(Direct Remote Procedure Call)以及executor、worker和task的关系及优化,以及如何...

    Storm配置项详解

    - **`drpc.port`**:设置Storm DRPC的服务端口,DRPC(Distributed RPC)允许外部系统调用Storm中定义的函数。 #### Supervisor 配置 - **`supervisor.slots.ports`**:定义supervisor上能够运行workers的端口列表...

    dRPC

    【dRPC】是一种分布式远程过程调用(Remote Procedure Call)框架,主要应用于Python编程环境。dRPC设计的目标是简化服务间的通信,提高系统的可扩展性和灵活性。作为一个轻量级的解决方案,它允许开发者以简单、...

    DRPC:简单的Discord RPC程序

    DRPC,全称为Discord Rich Presence Client,是一个用于实现Discord Rich Presence功能的简单程序。Discord Rich Presence是Discord平台提供的一种增强型状态显示功能,允许用户在游戏或应用程序中展示更丰富的交互...

    Storm入门教程 之Storm原理和概念详解

    Storm入门教程 之Storm原理和概念详解,出自Storm流计算从入门到精通之技术篇,Storm入门视频教程用到技术:Storm集群、Zookeeper集群等,涉及项目:网站PV、UV案例实战、其他案例; Storm视频教程亮点: 1、Storm...

Global site tag (gtag.js) - Google Analytics