推ter Storm: DRPC簡介

jopen 10年前發布 | 20K 次閱讀 Storm 分布式/云計算/大數據

轉自：http://xumingming.sinaapp.com/756/推ter-storm-drpc/

-------------------------

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://推ter.com。

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

</div> </div>

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

</div>
</div>
可以看出來， 我們需要做的事情非常的少。創建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();</pre> 

首先你創建一個LocalDRPC對象。 這個對象在進程內模擬一個DRPC服務器，跟LocalClusterLinearTopologyBuilder有單獨的方法來創建本地的topology和遠程的topology。在本地模式里面LocalDRPC對象不和任何端口綁定，所以我們的topology對象需要知道和誰交互。這就是為什么createLocalTopology方法接受一個LocalDRPC對象作為輸入的原因。


把topology啟動了之后，你就可以通過調用LocalDRPC對象的execute來調用RPC方法了。

遠程模式DRPC

在一個真是集群上面DRPC也是非常簡單的，有三個步驟:


啟動DRPC服務器

配置DRPC服務器的地址

提交DRPC topology到storm集群里面去。
</ul>
我們可以通過下面的storm腳本命令來啟動DRPC服務器:



幫助

bin/stormdrpc</div>
</div>
接著， 你需要讓你的storm集群知道你的DRPC服務器在哪里。DRPCSpout需要這個地址從而可以從DRPC服務器來接收函數調用。這個可以配置在storm.yaml或者通過代碼的方式配置在topology里面。通過storm.yaml配置是這樣的:


drpc.servers:

"drpc1.foo.com"
"drpc2.foo.com"</pre>


最后，你通過StormSubmitter對象來提交DRPC topology — 跟你提交其它topology沒有區別。如果要以遠程的方式運行上面的例子，用下面的代碼:

StormSubmitter.submitTopology(
"exclamation-drpc",
conf,
builder.createRemoteTopology()
);



我們用createRemoteTopology方法來創建運行在真實集群上的DRPC topology。


一個更復雜的例子
上面的DRPC例子只是為了介紹DRPC概念的一個簡單的例子。下面讓我們看一個復雜的、確實需要storm的并行計算能力的例子, 這個例子計算推ter上面一個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"));</pre><br />


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

}
}</pre>


當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服務器并且返回結果

</li>
LinearDRPCTopologyBuilder是利用storm的原語來構建高層抽象的很好的例子。
</ul>
高級特性

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





                    
                    
                         本文由用戶 jopen 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
                         轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
                         本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！
                         本文地址：http://www.baiduhome.net/lib/view/open1406168460476.html
                         Storm 分布式/云計算/大數據
                    

                
                
                    
                        相關經驗
                        
  Twitter Storm: DRPC簡介
   Storm集群的DRPC模式
   storm簡介
   Storm簡介
   storm 簡介及單機版安裝指南
                         
                    
                    
                        相關資訊
                        
  Twitter Storm 開源實時Hadoop簡介
   Apache Storm 的歷史及經驗教訓
   小藍單車推新款變速Bluegogo Pro，并推免押金服務
   Storm 并非完全適合所有實時應用
   Twitter已經用Heron替換了Storm
                         
                    
                    
                        相關文檔
                        
   Storm 簡介
    Storm高級原語
    storm設計開發規范-5
    淺析Storm流式計算
    Storm對于交通數據的處理
    Storm 應用場景
                         
                    

                    目錄