Storm分布式RPC

资料

Distributed RPC(分布式RPC)

下面是 storm的原文（http://storm.apache.org/documentation/Distributed-RPC.html）

The idea behind distributed RPC (DRPC) is to parallelize the computation of really intense functions on the fly using Storm. The Storm topology takes in as input a stream of function arguments, and it emits an output stream of the results for each of those function calls.
DRPC是为了真正实现在storm上高速计算的。Storm topology获取输入方法参数流，发射每个方法调用的结果数据流。

DRPC is not so much a feature of Storm as it is a pattern expressed from Storm’s primitives of streams, spouts, bolts, and topologies. DRPC could have been packaged as a separate library from Storm, but it’s so useful that it’s bundled with Storm.
DRPC并不是Storm的一个特性，不像Storm的streams，spouts，bolts和topology。DRPC可以被打包到一个独立于Storm的包，但是它对Storm是很有用的。

分布式RPC（distributed RPC，DRPC）用于对Storm上大量的函数调用进行并行计算过程。对于每一次函数调用，Storm集群上运行的拓扑接收调用函数的参数信息作为输入流，并将计算结果作为输出流发射出去。

DRPC本身算不上Storm的特性，它是通过Storm的基本元素：streams，spouts，bolts，topologies而衍生的一个模式。DRPC可以单独作为一个独立于Storm的库发布，但由于其重要性还是和Storm捆绑在了一起。

总体概述

Distributed RPC is coordinated by a “DRPC server” (Storm comes packaged with an implementation of this). The DRPC server coordinates receiving an RPC request, sending the request to the Storm topology, receiving the results from the Storm topology, and sending the results back to the waiting client. From a client’s perspective, a distributed RPC call looks just like a regular RPC call. For example, here’s how a client would compute the results for the “reach” function with the argument “http://twitter.com”:
分布式RPC是需要”DRPC server”配合的（storm有一个实现）。DRPC server协调接收RPC请求，发送请求到Storm topology, 接收Storm topology的结果，发送结果给等待的客户端。从客户端来看，一个分布式RPC调用就像一个regular RPC调用。例如这有一个客户端使用参数”http://twitter.com“计算”reach”方法结果：

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

The distributed RPC workflow looks like this:
DRPC工作流程如下：

The distributed RPC workflow looks like this:

Tasks in a topology

A client sends the DRPC

A client sends the DRPC server the name of the function to execute and the arguments to that function. The topology implementing that function uses a DRPCSpout to receive a function invocation stream from the DRPC server. Each function invocation is tagged with a unique id by the DRPC server. The topology then computes the result and at the end of the topology a bolt called ReturnResults connects to the DRPC server and gives it the result for the function invocation id. The DRPC server then uses the id to match up that result with which client is waiting, unblocks the waiting client, and sends it the result.
一个客户端发送DRPC server方法名和方法执行参数。topology实现该方法，用DRPCSpout接收方法调用数据流从DRPC server。每个方法调用会被DRPC server被打上唯一id标识。topology计算结果，在topology的结尾bolt上调用ReturnResults链接DRPC server，给它该方法调用id的结果。DRPC server用这个id匹配等待的客户端，解锁该等待客户端，发送结果给它。

DRPC通过DRPC Server来实现，DRPC Server的整体工作过程如下：

  1. 接收到一个RPC调用请求；
  2. 发送请求到Storm上的拓扑；
  3. 从Storm上接收计算结果；
  4. 将计算结果返回给客户端。

以上过程，在client客户端看来，一个DRPC调用看起来和一般的RPC调用没什么区别。下面代码是client通过DRPC调用“reach”函数，参数为“http://twitter.com”：

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

DRPC内部工作流程如下：

  1. Client向DRPC Server发送被调用执行的DRPC函数名称及参数。
  2. Storm上的topology通过DRPCSpout实现这一函数，从DPRC Server接收到函数调用流；
  3. DRPC Server会为每次函数调用生成唯一的id；
  4. Storm上运行的topology开始计算结果，最后通过一个ReturnResults的Bolt连接到DRPC Server，发送指定id的计算结果；
  5. DRPC Server通过使用之前为每个函数调用生成的id，将结果关联到对应的发起调用的client，将计算结果返回给client。

LinearDRPCTopologyBuilder

Storm提供了一个topology builder——LinearDRPCTopologyBuilder，它可以自动完成几乎所有的DRPC步骤。包括：

  1. 构建spout；
  2. 向DRPC Server返回结果；
  3. 为Bolt提供函数用于对tuples进行聚集。

下面是一个简单的例子，这个DRPC拓扑只是简单的在输入参数后追加“!”后返回：

public static class ExclaimBolt extends BaseBasicBolt { public void execute(Tuple tuple, BasicOutputCollector collector) { String input = tuple.getString(1); collector.emit(new Values(tuple.getValue(0), input + "!")); } 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函数的名称“exclamation”。一个DRPC Server可以协调多个函数，每个函数有不同的函数名称。拓扑中的第一个bolt的输入是两个字段：第一个是请求的id号；第二个是请求的参数。

LinearDRPCTopologyBuilder同时需要最后一个bolt发射一个包含两个字段的输出流：第一个字段是请求id；第二个字段是计算结果。因此，所有的中间tuples必须包含请求id作为第一个字段。

例子中，ExclaimBolt在输入tuple的第二个字段后面追加“!”，LinearDRPCTopologyBuilder负责处理其余的协调工作：与DRPC Server建立连接，发送结果给DRPC Server。

本地模式DRPC

DRPC可以以本地模式运行，下面的代码是如何在本地模式运行上面的例子：

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 Server，正如LocalCluster在本地模拟一个Storm集群一样。然后创建一个LocalCluster对象在本地模式下运行拓扑。LinearDRPCTopologyBuilder含有单独的方法用于创建本地拓扑和远程拓扑。

本地模式下，LocalDRPC并不绑定任何端口，因此Storm的拓扑需要了解要通讯的对象——这就是为什么createLocalTopology方法需要以LocalDRPC对象作为输入。

加载完拓扑之后，通过对LocalDRPC调用execute方法，就可以执行DRPC函数调用了。

远程模式DRPC

在实际的Storm集群上运行DRPC也一样很简单。只需完成以下步骤：

  1. 启动DRPC Server(s)；
  2. 配置DRPC Server(s)地址；
  3. 向Storm集群提交DRPC拓扑。

首先，通过storm脚本启动DRPC Server：

bin/storm drpc

然后，在Storm集群中配置DRPC Server地址，这就是DRPCSpout读取函数调用请求的地方。这一步的配置可以通过storm.yaml文件或者拓扑的配置来完成。通过storm.yaml文件的配置方式如下：

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

最后，通过StormSubmitter启动DRPC拓扑。为了以远程模式运行上面的例子，代码如下：

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

createRemoteTopology被用于为Storm集群创建合适的拓扑。

一个复杂的例子

上面的exclamation只是一个简单的DRPC例子。下面通过一个复杂的例子介绍如何在Storm集群内进行DRPC——计算Twitter上每个URL的到达度（reach），也就是每个URL暴露给的不同人的个数。

为了完成这一计算，需要完成以下步骤：

  1. 获取所有点选了（tweet）该URL的人；
  2. 获取步骤1中所有人的关注者（followers，粉丝）；
  3. 对所有关注者followers进行去重；
  4. 对步骤3中的关注者人数进行求和。

一个简单的URL到达度计算可能涉及成千上万次数据库调用以及数以百万的followers记录，计算量非常大。有了Storm，将很容易实现这一计算过程。单机上可能需要运行几分钟才能完成，在Storm集群上，即使是最难计算的URL也只需要几秒钟。

这个例子的代码在storm-starter：点击这里。这里是如何创建拓扑的代码：

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"));

拓扑的执行分为以下四步：

  1. GetTweeters：获取所有tweet了指定URL的用户列表，这个Bolt将输入流[id, url]转换成输出流[id, tweeter]，每个url元组被映射为多个tweeter元组。
  2. GetFollowers：获取步骤1中所有用户列表的followers，这个Bolt将输入流[id, twetter]转换成输出流[id, follower]，当某个人同时是多个人的关注者follower，而且这些人都tweet了指定的URL，那么将产生重复的follower元组。
  3. PartialUniquer：将所有followers按照follower id分组，使得同一个follower在同一个task中被处理。这个Bolt接收follower并进行去重计数。
  4. CountAggregator：从各个PartialUniquer中接收各部分的计数结果，累加后完成到达度计算。

下面是PartialUniquer这个Bolt的代码实现：

public class PartialUniquer extends BaseBatchBolt { BatchOutputCollector _collector; Object _id; Set<String> _followers = new HashSet<String>(); @Override public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) { _collector = collector; _id = id; } @Override public void execute(Tuple tuple) { _followers.add(tuple.getString(1)); } @Override public void finishBatch() { _collector.emit(new Values(_id, _followers.size())); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "partial-count")); } }

PartialUniquer通过继承BaseBatchBolt实现了IBatchBolt接口，batch bolt提供了API用于将一批tuples作为整体来处理。每个请求id会创建一个新的batch bolt实例，同时Storm负责这些实例的清理工作。

当PartialUniquer接收到一个follower元组时执行execute方法，将follower添加到请求id对应的HashSet集合中。

Batch bolt同时提供了finishBatch方法用于当这个task已经处理完所有的元组时调用。PartialUniquer发射一个包含当前task所处理的follower ids子集去重后个数的元组。

在内部实现上，CoordinatedBolt用于检测指定的bolt是否已经收到指定请求id的所有tuples元组。CoordinatedBolt使用direct streams管理实现这一协作过程。

拓扑的其他部分易于理解。到达度的每一步的计算过程都是并行进行的，通过DRPC实现也是非常容易的。

Non-linear DRPC拓扑

LinearDRPCTopologyBuilder只能处理“线性的”DRPC拓扑——正如到达度这样可以通过一系列步骤序列来完成的计算。不难想象，DRPC调用中包含有更复杂的带有分支和合并Bolt的拓扑。目前，必须自己直接使用CoordinatedBolt来完成这种非线性拓扑的计算。

LinearDRPCTopologyBuilder工作过程

  • DRPCSpout发射[args, return-info]，其中return-info包含DRPC Server的主机和端口号，以及DRPC Server为该次请求生成的唯一id号；
  • 构造一个Storm拓扑包含以下部分：
    • DRPCSpout
    • PrepareRequest(生成一个请求id，为return info创建一个流，为args创建一个流)
    • CoordinatedBolt wrappers以及direct groupings
    • JoinResult(将结果与return info拼接起来)
    • ReturnResult(连接到DRPC Server，返回结果)
  • LinearDRPCTopologyBuilder是建立在Storm基本元素之上的高层抽象。

高级进阶

KeyedFairBolt for weaving the processing of multiple requests at the same time

How to use CoordinatedBolt directly

  • KeyedFairBolt用于组织同一时刻多请求的处理过程；
  • 如何直接使用CoordinatedBolt。

发表评论