Chinaunix首页 | 论坛 | 博客
  • 博客访问: 447409
  • 博文数量: 80
  • 博客积分: 0
  • 博客等级: 民兵
  • 技术积分: 1916
  • 用 户 组: 普通用户
  • 注册时间: 2013-07-11 22:01
个人简介

从事实时计算多年,熟悉jstorm/spark/flink/kafka/rocketMq, 热衷于开源,希望在这里和前辈们一起学习与分享,得到长足的进步!邮箱:hustfxj@gmail.com 我的githup地址是:https://github.com/hustfxj。欢迎和大家一起交流探讨问题。

文章分类

全部博文(80)

文章存档

2017年(11)

2015年(3)

2014年(33)

2013年(33)

分类: 云计算

2014-07-09 14:42:43

原文地址:用storm做drpc真的很赞 作者:playmud

原创作品,允许转载,转载时请务必以超链接形式标明文章 原始出处 、作者信息和本声明。否则将追究法律责任。http://chenlx.blog.51cto.com/4096635/748348

分布式RPC

分布式RPC(DRPC)的真正目的是使用storm实时并行计算极端功能。Storm拓扑需要一个输入流作为函数参数,以一个输出流的形式发射每个函数调用的结果。
 
DRPC没有多少storm特性,因为它是从storm的原始流,spouts,bolts,拓扑来表达一个模式。DRPC没有单独打包,但它如此有用,以至于和storm捆绑在一起。
 
概述
分布式RPC通过“DRPC server”协调。DRPC服务器协调接收一个RPC请求,发送请求到storm拓扑,从storm拓扑接收结果,发送结果回等待的客户端。从一个客户端的角度来看,一个分布式RPC调用就像是一个常规的RPC调用。例如,一个客户端如何为带“http://twitter.com”参数的“reach”功能计算结果。
  1. DRPCClient client = new DRPCClient("drpc-host"3772); 
  2. String result = client.execute("reach""http://twitter.com"); 
分布式RPC工作流程如下:

客户端发送功能名称及功能所需参数到DRPC服务器去执行。图中的拓扑实现了此功能,它使用DRPCSpout从DRPC服务器接收功能调用流。每个功能调用通过DRPC服务器使用唯一ID标记,随后拓扑计算结果,在拓扑的最后,一个称之为“ReturnResults”的bolt连接到DRPC服务器,把结果交给这个功能调用(根据功能调用ID),DRPC服务器根据ID找到等待中的客户端,为等待中的客户端消除阻塞,并发送结果给客户端。

LinearDRPCTopologyBuilder
Storm有一个称之为LinearDRPCTopologyBuilder的拓扑Builder几乎自动完成DRPC所需的所有相关步骤。包括:
1.设置spout
2.返回结果给DRPC服务器
3.为bolt提供对一组元组的有限聚合功能
让我们看一个简单的例子。这是一个DRPC拓扑的实现,在输入参数的尾部追加“!”并返回:
  1. public static class ExclaimBolt implements IBasicBolt { 
  2.     public void prepare(Map conf, TopologyContext context) { 
  3.     } 
  4.  
  5.     public void execute(Tuple tuple, BasicOutputCollector collector) { 
  6.         String input = tuple.getString(1); 
  7.         collector.emit(new Values(tuple.getValue(0), input + "!")); 
  8.     } 
  9.  
  10.     public void cleanup() { 
  11.     } 
  12.  
  13.     public void declareOutputFields(OutputFieldsDeclarer declarer) { 
  14.         declarer.declare(new Fields("id""result")); 
  15.     } 
  16.  
  17.  
  18. public static void main(String[] args) throws Exception { 
  19.     LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation"); 
  20.     builder.addBolt(new ExclaimBolt(), 3); 
  21.     // ... 

如你所见,代码非常少。当创建时,你把这个拓扑的DRPC功能名称告诉storm。一个DRPC服务器可以协调许多功能,功能名称用于区别不同的功能,首先声明的bolt将接收一个输入的2-tuples,第一个字段是请求ID,第二个字段是请求参数。认为最后的bolt会发射一个输出流,该输出流包含[id, result]格式的2-tuples。最后,所有拓扑中间过程产生的元组(tuple)都包含请求id作为其第一个字段。

在这个例子中,ExclaimBolt只是简单地在元组的第二个字段尾部追加“!”字符。LinearDRPCTopologyBuilder处理其余的协调工作,包括连接DRPC服务器,发送最终结果。
 
本地模式DRPC
DRPC可以运行在本地模式。这是如何在本地模式运行上述例子:
  1. LocalDRPC drpc = new LocalDRPC(); 
  2. LocalCluster cluster = new LocalCluster(); 
  3.  
  4. cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc)); 
  5.  
  6. System.out.println("Results for 'hello':" + drpc.execute("exclamation""hello")); 
  7.  
  8. cluster.shutdown(); 
  9. drpc.shutdown(); 
首先你创建一个LocalDRPC对象。这个对象在进程内模拟一个DRPC服务器,就像在进程内模拟一个storm集群一样。然后你创建本地集群,在本地模式运行这个拓扑。创建本地拓扑和远程拓扑,LinearDRPCTopologyBuilder有不同的方法。在本地模式,LocalDRPC未绑定任何端口,拓扑也需要知道与哪个对象通讯,这是为什么createLocaclTopology方法需要接受LocalDRPC对象作为输入参数的原因。
 
载入拓扑后,你可以用LocalDRPC的execute方法执行DRPC调用。
 
远程模式DRPC
在实际的集群使用DRPC也很简单。有三个步骤:
1.      启动DRPC服务器
2.      配置DRPC服务器位置
3.      提交DRPC拓扑到storm集群
使用storm脚本启动DRPC服务器,和启动nimbus和ui一样:
  1. bin/storm drpc 
接下来,配置你的storm集群,让集群知道DRPC服务器的位置,这样DRPCSpout就知道从哪里读取功能调用。可以通过修改storm.yaml配置文件或拓扑配置完成配置DRPC服务器位置。修改storm.yaml配置文件如下所示:
  1. drpc.servers: 
  2.   - "drpc1.foo.com" 
  3.   - "drpc2.foo.com" 
最后,使用StormSubmitter启动DRPC拓扑,就像启动其它拓扑一样。在远程模式运行上述例子,代码如下所示:
  1. StormSubmitter.submitTopology("exclamation-drpc", conf, builder.createRemoteTopology()); 
createRemoteTopology方法用于在storm集群创建拓扑。
 
一个更完整的例子
这个exclaimation DRPC例子只是一个用来说明DRPC概念的玩具。让我们看一个更完整的例子,该例子是一个真正需要storm集群的并行计算的DRPC功能。我们将要看的例子是对twitter网站上的一个URL的接触用户进行统计。
一个URL的接触用户数是在twitter网站上接触一个URL的用户数,你需要以下4步:
1. 获取tweeted the URL的全部用户
2. 获取这些用户的全部追随者
3. 使追随者集合中的用户唯一
4. 统计唯一的用户数
一个单独的reach计算在计算期间涉及到数千数据库访问和数千万追随者记录。它是一个真正的耗时计算。正如你将要看到的,在storm上实现这个功能非常简单。在一台机器上,reach计算花费数分钟,在storm集群,最难计算reach的URL也只需数秒。
Storm-starter项目定义了一个reach样例,reach拓扑定义如下所示:
  1. LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach"); 
  2. builder.addBolt(new GetTweeters(), 3); 
  3. builder.addBolt(new GetFollowers(), 12
  4.         .shuffleGrouping(); 
  5. builder.addBolt(new PartialUniquer(), 6
  6.         .fieldsGrouping(new Fields("id""follower")); 
  7. builder.addBolt(new CountAggregator(), 2
  8.         .fieldsGrouping(new Fields("id")); 
这个拓扑以4个步骤的形式执行:

1. GetTweeters获取tweeted the URL的用户。它转换一个[id, url]形式的输入流到[id, tweeter]形式的输出流。每个url元组将映射到多个tweeter元组。

2. GetFollowers获取这些tweeter的追随者。它转换一个[id, tweeter]形式的输入流到[id, follower]形式的输出流。跨所有任务,当某人追随多个tweeter,这些tweeter又tweeted相同的URL时,这可能会得到重复的追随者。

按追随者ID对追随者数据流进行分组。同一的追随者去到同一的任务,因此每个PartialUniquer任务都接收到独立的相互独立的追随者集合。一旦收到请求ID用于它的所有追随者元组,它就发射追随者子集的唯一总数。
4. 最后,CountAggregator从每个PartialUniquer任务接收计数并对它们求和。
让我们来看看:
  1. public static class PartialUniquer implements IRichBolt, FinishedCallback { 
  2.     OutputCollector _collector; 
  3.     Map> _sets = new HashMap>(); 
  4.      
  5.     public void prepare(Map conf, TopologyContext context, OutputCollector collector) { 
  6.         _collector = collector; 
  7.     } 
  8.  
  9.     public void execute(Tuple tuple) { 
  10.         Object id = tuple.getValue(0); 
  11.         Set curr = _sets.get(id); 
  12.         if(curr==null) { 
  13.             curr = new HashSet(); 
  14.             _sets.put(id, curr); 
  15.         } 
  16.         curr.add(tuple.getString(1)); 
  17.         _collector.ack(tuple); 
  18.     } 
  19.  
  20.     public void cleanup() { 
  21.     } 
  22.  
  23.     public void finishedId(Object id) { 
  24.         Set curr = _sets.remove(id); 
  25.         int count; 
  26.         if(curr!=null) { 
  27.             count = curr.size(); 
  28.         } else { 
  29.             count = 0
  30.         } 
  31.         _collector.emit(new Values(id, count)); 
  32.     } 
  33.  
  34.     public void declareOutputFields(OutputFieldsDeclarer declarer) { 
  35.         declarer.declare(new Fields("id""partial-count")); 
  36.     } 
当PartialUniquer在exectue方法中接收一个follower元组时,它用一个内部HashMap添加它到与请求ID对应的集合。
PartialUniquer也实现了FinishedCallback接口,它告诉LinearDRPCTopologyBuilder,对于任意给定的请求ID,当它已收到所有指向它的元组时,请通知它。这个回调是finishedId方法。在这个回调中,PartialUniquer发射单一的元组,元组包含它的追随者子集的唯一总数。 

在底层,用于检测一个bolt何时收到该请求ID的所有元组。CoordinatedBolt使用direct stream管理协调。

其它的拓扑应该是不言自明。如你所见,reach计算的每一单步都是并行执行的,而且定义一个DRPC拓扑也非常简单。
Non-Linear DRPC拓扑
LinearDRPCTopologyBuilder仅处理“线性的”DRPC拓扑,计算以一连串步骤的形式表达(像reach)。不难想象某些功能将需要更复杂的拓扑结构,这些拓扑带有带分支和合并bolt。目前,要做到这一点,你需要直接使用CoordinateBolt。务必在邮件列表中谈谈你的非线性DRPC拓扑用例,写下DRPC拓扑更普遍的抽象结构。
LinearDRPCTopologyBuilder如何工作?

DRPCSpout发射[args, ],return-info是DRPC服务器的主机和端口,还有DRPC服务器生成的ID。

拓扑组成部分:
  • DRPCSpout
  • PrepareRequest(生成一个请求ID,创建一个返回信息流,一个参数流)
  • CoordinatedBolt包装器和直接分组
  • JoinResult(同返回信息一起连接结果)
  • ReturnResult(连接DRPC服务器并返回结果)
  • LinearDRPCTopologyBuilder是一个构建在Storm原语之上的高层次抽象的好例子。
高级
  • 同时编排处理多个请的KeyedFairBolt
  • 如何直接使用
 
 
英文:
阅读(1303) | 评论(0) | 转发(0) |
0

上一篇:storm简介

下一篇:storm基本概念总结

给主人留下些什么吧!~~