Storm專題一、Storm DRPC 分布式計算

本文譯自：https://storm.incubator.apache.org/documentation/Distributed-RPC.html

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方法的,方法的參數是: https://twitter.com。

java DRPCClient client = new DRPCClient("drpc-host", 3772); String result = client.execute("reach", "https://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的實現:幫助

publicstatic class ExclaimBolt implementsIBasicBolt {
    publicvoid prepare(Map conf, TopologyContext context) {
    }
 
    publicvoid execute(Tuple tuple, BasicOutputCollector collector) {
        String input = tuple.getString(1);
        collector.emit(newValues(tuple.getValue(0), input + "!"));
    }
 
    publicvoid cleanup() {
    }
 
    publicvoid declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(newFields("id","result"));
    }
 
}
 
publicstatic void main(String[] args) throwsException {
    LinearDRPCTopologyBuilder builder
        =newLinearDRPCTopologyBuilder("exclamation");
    builder.addBolt(newExclaimBolt(), 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 = newLocalDRPC();
LocalCluster cluster = newLocalCluster();
 
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/stormdrpc

接著，你需要讓你的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
    =newLinearDRPCTopologyBuilder("reach");
builder.addBolt(newGetTweeters(), 3);
builder.addBolt(newGetFollowers(), 12)
        .shuffleGrouping();
builder.addBolt(newPartialUniquer(), 6)
        .fieldsGrouping(newFields("id","follower"));
builder.addBolt(newCountAggregator(), 2)
        .fieldsGrouping(newFields("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的實現:

publicstatic class PartialUniquer
        implementsIRichBolt, FinishedCallback {
 
    OutputCollector _collector;
    Map<Object, Set<String>> _sets
                =newHashMap<Object, Set<String>>();
 
    publicvoid prepare(Map conf,
                        TopologyContext context,
                        OutputCollector collector) {
        _collector = collector;
    }
 
    publicvoid execute(Tuple tuple) {
        Object id = tuple.getValue(0);
        Set<String> curr = _sets.get(id);
        if(curr==null) {
            curr = newHashSet<String>();
            _sets.put(id, curr);
        }
        curr.add(tuple.getString(1));
        _collector.ack(tuple);
    }
 
    publicvoid cleanup() {
    }
 
    publicvoid finishedId(Object id) {
        Set<String> curr = _sets.remove(id);
        intcount;
        if(curr!=null) {
            count = curr.size();
        }else{
            count = 0;
        }
        _collector.emit(newValues(id, count));
    }
 
    publicvoid declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(newFields("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

最後更新：2017-04-04 07:32:00

Storm專題一、Storm DRPC 分布式計算

概覽

LinearDRPCTopologyBuilder

本地模式DRPC

遠程模式DRPC

一個更複雜的例子

非線性DRPC Topology

LinearDRPCTopologyBuilder的工作原理

高級特性

上一篇：機房收費係統之結賬與報表

下一篇： Oracle性能優化學習筆記之WHERE子句中的連接順序

相關內容

熱門內容

最新內容

Storm專題一、Storm DRPC 分布式計算

概覽

LinearDRPCTopologyBuilder

本地模式DRPC

遠程模式DRPC

一個更複雜的例子

非線性DRPC Topology

LinearDRPCTopologyBuilder的工作原理

高級特性

上一篇： 機房收費係統之結賬與報表

下一篇： Oracle性能優化學習筆記之WHERE子句中的連接順序

相關內容

熱門內容

最新內容

上一篇：機房收費係統之結賬與報表