【流式计算】Twitter Storm: DRPC简介

作者: xumingming | 可以转载, 但必须以超链接形式标明文章原始出处和作者信息及版权声明

网址: http://xumingming.sinaapp.com/756/twitter-storm-drpc/

本文翻译自：https://github.com/nathanmarz/storm/wiki/Distributed-RPC。

Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU intensive的计算。DRPC的storm topology以函数的参数流作为输入，而把这些函数调用的返回值作为topology的输出流。

DRPC其实不能算是storm本身的一个特性， 它是通过组合storm的原语spout，bolt， topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的， 但是DRPC实在是太有用了，所以我们我们把它和storm捆绑在一起。

概览

Distributed RPC是由一个”DPRC Server”协调的(storm自带了一个实现)。DRPC服务器协调1) 接收一个RPC请求。2) 发送请求到storm topology 3) 从storm topology接收结果。4) 把结果发回给等待的客户端。从客户端的角度来看一个DRPC调用跟一个普通的RPC调用没有任何区别。比如下面是客户端如何调用RPC： reach方法的,方法的参数是: http://twitter.com。

帮助

1 2 3

DRPCClient client = new DRPCClient( "drpc-host" , 3772 ); String result = client.execute( "reach" , "http://twitter.com" );

DRPC的工作流大致是这样的:

客户端给DRPC服务器发送要执行的方法的名字，以及这个方法的参数。实现了这个函数的topology使用

DRPCSpout

从DRPC服务器接收函数调用流。每个函数调用被DRPC服务器标记了一个唯一的id。 这个topology然后计算结果，在topology的最后一个叫做

ReturnResults

的bolt会连接到DRPC服务器，并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上，唤醒这个客户端并且把结果发送给它。

LinearDRPCTopologyBuilder

Storm自带了一个称作LinearDRPCTopologyBuilder的topology builder, 它把实现DRPC的几乎所有步骤都自动化了。这些步骤包括:

• 设置spout
• 把结果返回给DRPC服务器
• 给bolt提供有限聚合几组tuples的能力

让我们看一个简单的例子。下面是一个把输入参数后面添加一个”!”的DRPC topology的实现:

帮助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

public static class ExclaimBolt implements IBasicBolt { public void prepare(Map conf, TopologyContext context) { } public void execute(Tuple tuple, BasicOutputCollector collector) { String input = tuple.getString( 1 ); collector.emit( new Values(tuple.getValue( ), input + "!" )); } public void cleanup() { } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare( new Fields( "id" , "result" )); } } public static void main(String[] args) throws Exception { LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder( "exclamation" ); builder.addBolt( new ExclaimBolt(), 3 ); // ... }

可以看出来， 我们需要做的事情非常的少。创建

LinearDRPCTopologyBuilder

的时候，你需要告诉它你要实现的DRPC函数的名字。一个DRPC服务器可以协调很多函数，函数与函数之间靠函数名字来区分。你声明的第一个bolt会接收两维tuple，tuple的第一个field是request-id，第二个field是这个请求的参数。

LinearDRPCTopologyBuilder

同时要求我们topology的最后一个bolt发射一个二维tuple： 第一个field是request-id， 第二个field是这个函数的结果。最后所有中间tuple的第一个field必须是request-id。

在这里例子里面

ExclaimBolt

简单地在输入tuple的第二个field后面再添加一个”!”，其余的事情都由

LinearDRPCTopologyBuilder

帮我们搞定：链接到DRPC服务器，并且把结果发回。

本地模式DRPC

DRPC可以以本地模式运行。下面就是以本地模式运行上面例子的代码:

帮助

01 02 03 04 05 06 07 08 09 10 11 12 13 14

LocalDRPC drpc = new LocalDRPC(); LocalCluster cluster = new LocalCluster(); cluster.submitTopology( "drpc-demo" , conf, builder.createLocalTopology(drpc) ); System.out.println( "Results for 'hello':" + drpc.execute( "exclamation" , "hello" )); cluster.shutdown(); drpc.shutdown();

首先你创建一个

LocalDRPC

对象。 这个对象在进程内模拟一个DRPC服务器，跟

LocalClusterLinearTopologyBuilder

有单独的方法来创建本地的topology和远程的topology。在本地模式里面

LocalDRPC

对象不和任何端口绑定，所以我们的topology对象需要知道和谁交互。这就是为什么

createLocalTopology

方法接受一个

LocalDRPC

对象作为输入的原因。

把topology启动了之后，你就可以通过调用

LocalDRPC

对象的

execute

来调用RPC方法了。

远程模式DRPC

在一个真是集群上面DRPC也是非常简单的，有三个步骤:

• 启动DRPC服务器
• 配置DRPC服务器的地址
• 提交DRPC topology到storm集群里面去。

我们可以通过下面的

storm

脚本命令来启动DRPC服务器:

帮助

1	bin /storm drpc

接着， 你需要让你的storm集群知道你的DRPC服务器在哪里。

DRPCSpout

需要这个地址从而可以从DRPC服务器来接收函数调用。这个可以配置在

storm.yaml

或者通过代码的方式配置在topology里面。通过

storm.yaml

配置是这样的:

帮助

1 2 3

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

最后，你通过

StormSubmitter

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

帮助

1 2 3 4 5

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

我们用

createRemoteTopology

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

一个更复杂的例子

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

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

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

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

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

帮助

1 2 3 4 5 6 7 8 9

LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder( "reach" ); builder.addBolt( new GetTweeters(), 3 ); builder.addBolt( new GetFollowers(), 12 ) .shuffleGrouping(); builder.addBolt( new PartialUniquer(), 6 ) .fieldsGrouping( new Fields( "id" , "follower" )); builder.addBolt( new CountAggregator(), 2 ) .fieldsGrouping( new Fields( "id" ));

这个topology分四步执行:

• GetTweeters

  获取所发微薄里面包含制定URL的所有用户。它接收输入流:

  [id, url]

  , 它输出:

  [id, tweeter]

  . 没一个URL tuple会对应到很多

  tweeter

  tuple。

• GetFollowers

  获取这些tweeter的粉丝。它接收输入流:

  [id, tweeter]

  , 它输出:

  [id, follower]

• PartialUniquer

  通过粉丝的id来group粉丝。这使得相同的分析会被引导到统一个task。因此不同的task接收到的粉丝是不同的 — 从而起到去重的作用。它的输出流：

  [id, count]

  即输出这个task上统计的粉丝个数。
• 最后，

  CountAggregator

  接收到所有的局部数量， 把它们加起来就算出了我们要的reach值。

我们来看一下

PartialUniquer

的实现:

帮助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42

public static class PartialUniquer implements IRichBolt, FinishedCallback { OutputCollector _collector; Map<Object, Set<String>> _sets = new HashMap<Object, Set<String>>(); public void prepare(Map conf, TopologyContext context, OutputCollector collector) { _collector = collector; } public void execute(Tuple tuple) { Object id = tuple.getValue( ); Set<String> curr = _sets.get(id); if (curr== null ) { curr = new HashSet<String>(); _sets.put(id, curr); } curr.add(tuple.getString( 1 )); _collector.ack(tuple); } public void cleanup() { } public void finishedId(Object id) { Set<String> curr = _sets.remove(id); int count; if (curr!= null ) { count = curr.size(); } else { count = ; } _collector.emit( new Values(id, count)); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare( new Fields( "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