Storm: DRPC简介

爱被打了一巴掌 2022-06-03 01:08 199阅读 0赞

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。

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

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

[![drpc-workflow][]][drpc-workflow 1]

客户端给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的实现:

![复制代码][copycode.gif]

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);
        // ...
    }

![复制代码][copycode.gif]

可以看出来， 我们需要做的事情非常的少。创建 `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可以以本地模式运行。下面就是以本地模式运行上面例子的代码:

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

首先你创建一个 `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的例子可以在 [这里][Link 1] 找到(storm-starter)。reach topology是这样定义的:

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

这个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` 的实现:

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

当 `PartialUniquer` 在 `execute` 方法里面接收到一个 `粉丝tuple` 的时候， 它把这个tuple添加到当前request-id对应的 `Set `里面去。

`PartialUniquer` 同时也实现了 `FinishedCallback` 接口, 实现这个接口是告诉 `LinearDRPCTopologyBuilder` 它想在接收到某个request-id的所有tuple之后得到通知，回调函数则是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

一个更复杂的例子的全部代码

![复制代码][copycode.gif]

/**
     * Licensed to the Apache Software Foundation (ASF) under one
     * or more contributor license agreements.  See the NOTICE file
     * distributed with this work for additional information
     * regarding copyright ownership.  The ASF licenses this file
     * to you under the Apache License, Version 2.0 (the
     * "License"); you may not use this file except in compliance
     * with the License.  You may obtain a copy of the License at
     *
     * http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package cn.ljh.storm.drpc;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.LocalDRPC;
    import org.apache.storm.StormSubmitter;
    import org.apache.storm.coordination.BatchOutputCollector;
    import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.BasicOutputCollector;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.topology.base.BaseBasicBolt;
    import org.apache.storm.topology.base.BaseBatchBolt;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Tuple;
    import org.apache.storm.tuple.Values;
    
    import java.util.*;
    
    /**
     * This is a good example of doing complex Distributed RPC on top of Storm. This program creates a topology that can
     * compute the reach for any URL on Twitter in realtime by parallelizing the whole computation.
     * <p/>
     * Reach is the number of unique people exposed to a URL on Twitter. To compute reach, you have to get all the people
     * who tweeted the URL, get all the followers of all those people, unique that set of followers, and then count the
     * unique set. It's an intense computation that can involve thousands of database calls and tens of millions of follower
     * records.
     * <p/>
     * This Storm topology does every piece of that computation in parallel, turning what would be a computation that takes
     * minutes on a single machine into one that takes just a couple seconds.
     * <p/>
     * For the purposes of demonstration, this topology replaces the use of actual DBs with in-memory hashmaps.
     *
     * @see <a href="http://storm.apache.org/documentation/Distributed-RPC.html">Distributed RPC</a>
     */
    public class ReachTopology {
      public static Map<String, List<String>> TWEETERS_DB = new HashMap<String, List<String>>() {
        {
        put("foo.com/blog/1", Arrays.asList("sally", "bob", "tim", "george", "nathan"));
        put("engineering.twitter.com/blog/5", Arrays.asList("adam", "david", "sally", "nathan"));
        put("tech.backtype.com/blog/123", Arrays.asList("tim", "mike", "john"));
      }};
    
      public static Map<String, List<String>> FOLLOWERS_DB = new HashMap<String, List<String>>() {
        {
        put("sally", Arrays.asList("bob", "tim", "alice", "adam", "jim", "chris", "jai"));
        put("bob", Arrays.asList("sally", "nathan", "jim", "mary", "david", "vivian"));
        put("tim", Arrays.asList("alex"));
        put("nathan", Arrays.asList("sally", "bob", "adam", "harry", "chris", "vivian", "emily", "jordan"));
        put("adam", Arrays.asList("david", "carissa"));
        put("mike", Arrays.asList("john", "bob"));
        put("john", Arrays.asList("alice", "nathan", "jim", "mike", "bob"));
      }};
    
      public static class GetTweeters extends BaseBasicBolt {
        public void execute(Tuple tuple, BasicOutputCollector collector) {
          Object id = tuple.getValue(0);
          String url = tuple.getString(1);
          List<String> tweeters = TWEETERS_DB.get(url);
          if (tweeters != null) {
            for (String tweeter : tweeters) {
              collector.emit(new Values(id, tweeter));
            }
          }
        }
    
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("id", "tweeter"));
        }
      }
    
      public static class GetFollowers extends BaseBasicBolt {
        public void execute(Tuple tuple, BasicOutputCollector collector) {
          Object id = tuple.getValue(0);
          String tweeter = tuple.getString(1);
          List<String> followers = FOLLOWERS_DB.get(tweeter);
          if (followers != null) {
            for (String follower : followers) {
              collector.emit(new Values(id, follower));
            }
          }
        }
    
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("id", "follower"));
        }
      }
    
      public static class PartialUniquer extends BaseBatchBolt {
        BatchOutputCollector _collector;
        Object _id;
        Set<String> _followers = new HashSet<String>();
    
        public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
          _collector = collector;
          _id = id;
        }
    
        public void execute(Tuple tuple) {
          //利用set的特性来去重。
          _followers.add(tuple.getString(1));
        }
    
        public void finishBatch() {
          //同一个task处理完了相同id的tuple之后调用。
          _collector.emit(new Values(_id, _followers.size()));
        }
    
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("id", "partial-count"));
        }
      }
    
      public static class CountAggregator extends BaseBatchBolt {
        BatchOutputCollector _collector;
        Object _id;
        int _count = 0;
    
        public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
          _collector = collector;
          _id = id;
        }
    
        public void execute(Tuple tuple) {
          _count += tuple.getInteger(1);
        }
    
        public void finishBatch() {
          _collector.emit(new Values(_id, _count));
        }
    
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("id", "reach"));
        }
      }
    
      public static LinearDRPCTopologyBuilder construct() {
        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach");
        builder.addBolt(new GetTweeters(), 1);
        builder.addBolt(new GetFollowers(), 1).shuffleGrouping();
        builder.addBolt(new PartialUniquer(), 2).fieldsGrouping(new Fields("id", "follower"));
        builder.addBolt(new CountAggregator(), 1).fieldsGrouping(new Fields("id"));
        return builder;
      }
    
      public static void main(String[] args) throws Exception {
        LinearDRPCTopologyBuilder builder = construct();
    
    
        Config conf = new Config();
    
        if (args == null || args.length == 0) {
          conf.setMaxTaskParallelism(3);
          LocalDRPC drpc = new LocalDRPC();
          LocalCluster cluster = new LocalCluster();
          cluster.submitTopology("reach-drpc", conf, builder.createLocalTopology(drpc));
    
          String[] urlsToTry = new String[]{ "foo.com/blog/1", "engineering.twitter.com/blog/5", "notaurl.com" };
          for (String url : urlsToTry) {
            System.out.println("Reach of " + url + ": " + drpc.execute("reach", url));
          }
    
          cluster.shutdown();
          drpc.shutdown();
        }
        else {
          conf.setNumWorkers(6);
          StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createRemoteTopology());
        }
      }
    }

![复制代码][copycode.gif]

storm命令topoloy提交

storm jar /home/test/storm-helloworld-0.0.1-SNAPSHOT-jar-with-dependencies.jar cn.ljh.storm.drpc.ReachTopology ReachTopology

客户端代码

![复制代码][copycode.gif]

package cn.ljh.storm.drpc;
    
    import org.apache.storm.Config;
    import org.apache.storm.utils.DRPCClient;
    
    public class DRPCReachClient {
        public static void main(String[] args) throws Exception {
            Config conf = new Config();
            conf.setDebug(false);
            conf.put("storm.thrift.transport", "org.apache.storm.security.auth.SimpleTransportPlugin");
            conf.put(Config.STORM_NIMBUS_RETRY_TIMES, 3);
            conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL, 10);
            conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL_CEILING, 20);
            conf.put(Config.DRPC_MAX_BUFFER_SIZE, 1048576);
            DRPCClient drpcClient = new DRPCClient(conf, "192.168.137.180", 3772);
            System.out.println(drpcClient.execute("reach", "foo.com/blog/1"));
        }
    }

[drpc-workflow]: http://img1.tuicool.com/7VR3qa.png!web
[drpc-workflow 1]: http://xumingming-wordpress.stor.sinaapp.com/uploads/2012/02/drpc-workflow.png
[LinearDRPCTopologyBuilder]: http://nathanmarz.github.com/storm/doc/backtype/storm/drpc/LinearDRPCTopologyBuilder.html
[copycode.gif]: /images/20220603/14bad4a5c42e4aeea1d9ba76ca1a8eb0.png
[Link 1]: https://github.com/nathanmarz/storm-starter/blob/master/src/jvm/storm/starter/ReachTopology.java