推ter Storm: DRPC简介

bin/stormdrpc

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

drpc.servers:    - "drpc1.foo.com"    - "drpc2.foo.com"

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

StormSubmitter.submitTopology(      "exclamation-drpc",      conf,      builder.createRemoteTopology()  );

我们用createRemoteTopology方法来创建运行在真实集群上的DRPC topology。

一个更复杂的例子

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

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

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

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

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

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的实现:

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服务器并且返回结果
  </li>
• LinearDRPCTopologyBuilder是利用storm的原语来构建高层抽象的很好的例子。
</ul>

高级特性

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