Storm常见模式——分布式RPC

Storm常见模式——分布式RPC

本文翻译自：https://github.com/nathanmarz/storm/wiki/Distributed-RPC，做为学习Storm DRPC的资料，转载必须以超连接形式标明文章原始出处及本文翻译连接。

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

DRPC自己算不上Storm的特性，它是经过Storm的基本元素：streams，spouts，bolts，topologies而衍生的一个模式。DRPC能够单独做为一个独立于Storm的库发布，但因为其重要性仍是和Storm捆绑在了一块儿。

整体概述

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用于组织同一时刻多请求的处理过程；

• 如何直接使用CoordinatedBolt。