一个更复杂的例子
上面的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
- 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会对应到很多tweetertuple。GetFollowers获取这些tweeter的粉丝。它接收输入流:[id, tweeter], 它输出:[id, follower]PartialUniquer通过粉丝的id来group粉丝。这使得相同的分析会被引导到统一个task。因此不同的task接收到的粉丝是不同的 — 从而起到去重的作用。它的输出流：[id, count]即输出这个task上统计的粉丝个数。- 最后，
CountAggregator接收到所有的局部数量， 把它们加起来就算出了我们要的reach值。
我们来看一下PartialUniquer的实现:
[java] view plaincopy
- 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的原语来构建高层抽象的很好的例子。
高级特性
- 如何利用KeyedFairBolt来同时处理多个请求
- 如何直接使用CoordinatedBolt