首页 > 分布式 > Storm > Storm高级原语(二) — DRPC
2014
04-25

Storm高级原语(二) — DRPC

本文翻译自Apache Storm主页上的Distributed-RPC 介绍一文,同时参考徐明明博客。

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

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

概览

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

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

DRPC的工作流大致是这样的(重要☆):

How DRPC works

客户端给DRPC服务器发送要执行的函数(function)的名字,以及这个函数的参数。实现了这个函数的topology使用DRPCSpout从DRPC服务器接收函数调用流,每个函数调用被DRPC服务器标记了一个唯一的id。 这个topology然后计算结果,在topology的最后,一个叫做ReturnResults的bolt会连接到DRPC服务器,并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上,唤醒这个客户端并且把结果发送给它。

LinearDRPCTopologyBuilder

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

1、设置spout

2、把结果返回给DRPC服务器

3、给bolt提供有限聚合几组tuples的能力

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

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函数(DRPC function)的名字。一个DRPC服务器可以协调很多函数,函数与函数之间靠函数名字来区分。你声明的第一个bolt会接收一个两维tuple,tuple的第一个字段是request-id,第二个字段是这个请求的参数。LinearDRPCTopologyBuilder同时要求我们topology的最后一个bolt发送一个形如[id, result]的二维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服务器(这很类似于LocalCluster在进程内模拟一个Storm集群),然后创建LocalCluster对象在本地模式运行topology。LinearTopologyBuilder有单独的方法来创建本地的topology和远程的topology。在本地模式里面LocalDRPC对象不和任何端口绑定,所以我们的topology对象需要知道和谁交互,这就是为什么createLocalTopology方法接受一个LocalDRPC对象作为输入的原因。

把topology启动了之后,你就可以通过调用LocalDRPC对象的execute来调用RPC方法了。

远程模式DRPC

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

1、启动DRPC服务器

2、配置DRPC服务器的地址

3、提交DRPC topology到storm集群里面去。

我们可以通过“bin/storm drpc”命令来启动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值。

一个URL的reach值是该URL对应的推文能到达(reach)的用户数量,要计算一个URL的reach值,我们需要:

1、获取所有推文里面包含这个URL的人(转发过该URL的人)

2、获取这些人的粉丝

3、把这些粉丝去重

4、获取这些去重之后的粉丝个数 — 这就是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分四步执行:

1、GetTweeters获取转发该推文的所有用户。它接收输入流: [id, url],它输出:[id, tweeter]. 每个URL tuple会对应到很多tweeter tuple。

2、GetFollowers 获取这些转发者(tweeter)的粉丝。它接收输入流: [id, tweeter], 它输出: [id, follower]。当然,当某人关注的多个人都转发了同一条推文时,follower tuple会存在重复,这就需要下一步的去重。

3、PartialUniquer 通过粉丝的id来group粉丝,这使得相同的粉丝会被引导到同一个task。因此不同的task接收到的粉丝是不同的 — 从而起到去重的作用。它的输出流:[id, count] 即输出这个task上统计的粉丝个数。

4、最后,CountAggregator 接收到所有的局部数量, 把它们加起来就算出了我们要的reach值。

我们来看一下PartialUniquer的实现:

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在execute方法里面接收到一个粉丝tuple的时候, 它把这个tuple添加到当前request-id对应的Set里面去(利用Set元素不重复的特点进行去重)。

PartialUniquer继承了BaseBatchBolt类。对于每个request-id,创建一个相应batch bolt的实例,并且Storm会在合适的时候清理这些实例。batch bolt提供了finishBatch的方法,该方法将在这个batch中的所有tuple被处理完之后调用。PartialUniquer仅发送一个tuple,包含当前这个request-id在这个task上的粉丝数量。

LinearDRPCTopologyBuilder的工作原理

1、DRPCSpout发射tuple: [args, return-info]。 return-info包含DRPC服务器的主机地址、端口以及当前请求的request-id(DRPC服务器生成)

2、DRPC Topology包含以下元素:

  • DRPCSpout
  • PrepareRequest(生成request-id, return info以及args)
  • CoordinatedBolt
  • JoinResult — 通过return info组合结果
  • ReturnResult — 连接到DRPC服务器并且返回结果

3、LinearDRPCTopologyBuilder是利用storm的原语来构建高层抽象的很好的例子。


留下一个回复

你的email不会被公开。