【流式計算】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。

幫助

1 2 3

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的實作:

幫助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

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

幫助

01 02 03 04 05 06 07 08 09 10 11 12 13 14

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

幫助

1	bin /storm drpc

接着， 你需要讓你的storm叢集知道你的DRPC伺服器在哪裡。

DRPCSpout

需要這個位址進而可以從DRPC伺服器來接收函數調用。這個可以配置在

storm.yaml

或者通過代碼的方式配置在topology裡面。通過

storm.yaml

配置是這樣的:

幫助

1 2 3

drpc.servers: - "drpc1.foo.com" - "drpc2.foo.com"

最後，你通過

StormSubmitter

對象來送出DRPC topology — 跟你送出其它topology沒有差別。如果要以遠端的方式運作上面的例子，用下面的代碼:

幫助

1 2 3 4 5

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是這樣定義的:

幫助

1 2 3 4 5 6 7 8 9

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

的實作:

幫助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42

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( ); 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 = ; } _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的原語來建構高層抽象的很好的例子。

進階特性

• 如何利用KeyedFairBolt來同時處理多個請求
• 如何直接使用CoordinatedBolt