首页 > 代码库 > storm实时分析-示例2

storm实时分析-示例2

2024-07-30 23:56:28   223人阅读

一个更复杂的例子

上面的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. }  

 

 

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

高级特性

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

storm实时分析-示例2


联系
我们