码迷,mamicode.com
首页 > 其他好文 > 详细

storm实时分析-示例2

时间:2014-11-03 22:22:27      阅读:219      评论:0      收藏:0      [点我收藏+]

标签:blog   http   ar   使用   java   sp   strong   数据   div   

一个更复杂的例子

上面的DRPC例子只是为了介绍DRPC概念的一个简单的例子。下面让我们看一个复杂的、确实需要storm的并行计算能力的例子, 这个例子计算twitter上面一个url的reach值。

首先介绍一下什么是reach值,要计算一个URL的reach值,我们需要:

  • 获取所有微薄里面包含这个URL的人
  • 获取这些人的粉丝
  • 把这些粉丝去重
  • 获取这些去重之后的粉丝个数 — 这就是reach

一个简单的reach计算可能会有成千上万个数据库调用,并且可能设计到百万数量级的微薄用户。这个确实可以说是CPU intensive的计算了。你会看到的是,在storm上面来实现这个是非常非常的简单。在单台机器上面, 一个reach计算可能需要花费几分钟。而在一个storm集群里面,即时是最男的URL, 也只需要几秒。

一个reach topolgoy的例子可以在这里找到(storm-starter)。reach topology是这样定义的:

 

[java] view plaincopy
 
  1. LinearDRPCTopologyBuilder builder  
  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的实现:

 

[java] view plaincopy
 
  1. publicstatic class PartialUniquer  
  2.         implementsIRichBolt, FinishedCallback {  
  3.    
  4.     OutputCollector _collector;  
  5.     Map<Object, Set<String>> _sets  
  6.                 =newHashMap<Object, Set<String>>();  
  7.    
  8.     publicvoid prepare(Map conf,  
  9.                         TopologyContext context,  
  10.                         OutputCollector collector) {  
  11.         _collector = collector;  
  12.     }  
  13.    
  14.     publicvoid execute(Tuple tuple) {  
  15.         Object id = tuple.getValue(0);  
  16.         Set<String> curr = _sets.get(id);  
  17.         if(curr==null) {  
  18.             curr = newHashSet<String>();  
  19.             _sets.put(id, curr);  
  20.         }  
  21.         curr.add(tuple.getString(1));  
  22.         _collector.ack(tuple);  
  23.     }  
  24.    
  25.     publicvoid cleanup() {  
  26.     }  
  27.    
  28.     publicvoid finishedId(Object id) {  
  29.         Set<String> curr = _sets.remove(id);  
  30.         intcount;  
  31.         if(curr!=null) {  
  32.             count = curr.size();  
  33.         }else{  
  34.             count = 0;  
  35.         }  
  36.         _collector.emit(newValues(id, count));  
  37.     }  
  38.    
  39.     publicvoid declareOutputFields(OutputFieldsDeclarer declarer) {  
  40.         declarer.declare(newFields("id","partial-count"));  
  41.     }  
  42. }  

 

 

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的原语来构建高层抽象的很好的例子。

高级特性

    • 如何利用KeyedFairBolt来同时处理多个请求
    • 如何直接使用CoordinatedBolt

storm实时分析-示例2

标签:blog   http   ar   使用   java   sp   strong   数据   div   

原文地址:http://www.cnblogs.com/muzhongjiang/p/4072302.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!