Storm DRPC 介绍（2）

问题导读

1.LinearDRPCTopologyBuilder的作用是什么？
2.远程模式DRPC与本地DRPC有什么不同之处？
3.例子topology分几步执行？

LinearDRPCTopologyBuilder

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

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

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

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(0), 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可以以本地模式运行。下面就是以本地模式运行上面例子的代码:

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服务器:

bin/storm drpc
复制代码

接着，你需要让你的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的并行计算能力的例子, 这个例子计算twitter上面一个url的reach值。

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

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

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

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

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会对应到很多tweetertuple。
GetFollowers 获取这些tweeter的粉丝。它接收输入流: [id, tweeter], 它输出: [id, follower]
PartialUniquer 通过粉丝的id来group粉丝。这使得相同的分析会被引导到统一个task。因此不同的task接收到的粉丝是不同的 — 从而起到去重的作用。它的输出流：[id, count] 即输出这个task上统计的粉丝个数。
最后，CountAggregator 接收到所有的局部数量，把它们加起来就算出了我们要的reach值。

我们来看一下PartialUniquer的实现:

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(0);
        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 = 0;
        }
        _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的原语来构建高层抽象的很好的例子。

高级特性