Storm入門（十二）Twitter Storm: DRPC簡介

做者: xumingming | 能夠轉載, 但必須以超連接形式標明文章原始出處和做者信息及版權聲明
網址: http://xumingming.sinaapp.com/756/twitter-storm-drpc/

本文翻譯自： https://github.com/nathanmarz/storm/wiki/Distributed-RPC 。

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服務器發送要執行的方法的名字，以及這個方法的參數。實現了這個函數的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的實現:

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會對應到不少 tweeter tuple。
• 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以後獲得通知，回調函數則是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

一個更復雜的例子的所有代碼

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

 

storm命令topoloy提交

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

客戶端代碼

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