Twitter Storm: storm的一些常見模式

發表于  2011 年 10 月 07 日  由  xumingming 作者:  xumingming | 可以轉載, 但必須以超連結形式标明文章原始出處和作者資訊及版權聲明

網址:  http://xumingming.sinaapp.com/189/twitter-storm-storm的一些常見模式/

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

這篇文章列舉出了storm topology裡面的一些常見模式：

• 流聚合(stream join)
• 批處理(Batching)
• BasicBolt
• 記憶體内緩存 + fields grouping 組合
• 計算top N
• 用TimeCacheMap來高效地儲存一個最近被更新的對象的緩存
• 分布式RPC: CoordinatedBolt和KeyedFairBolt

流聚合(stream join)

流聚合把兩個或者多個資料流聚合成一個資料流 — 基于一些共同的tuple字段。流聚合和SQL裡面table join很像，隻是table join的輸入是有限的，并且join的語義是非常明确的。而流聚合的語義是不明确的并且輸入流是無限的。

流類型的聚合類型跟具體的應用是有關了。一些應用把兩個流發出的所有的tuple都聚合起來 — 不管多長時間；而另外一些應用則隻會聚合一些特定的tuple。而另外一些應用的聚合邏輯又可能完全不一樣。而這些聚合類型裡面最常見的類型是把所有的輸入流進行一樣的劃分， 這個在storm裡面用fields grouping在相同字段上進行grouping就可以了，比如:

幫助

1 2 3 4

builder.setBolt( 5 , new MyJoiner(), parallelism) .fieldsGrouping( 1 , new Fields( "field1" , "field2" )) .fieldsGrouping( 2 , new Fields( "field1" , "field2" )) .fieldsGrouping( 3 , new Fields( "field1" , "field2" ));

當然，不同的資料流的“相同”字段可以有不一樣的名字。

批處理(Batching)

有時候為了性能或者一些别的原因， 你可能想把一組tuple一起處理， 而不是一個個單獨處理。比如，你可能想批量更新資料庫。

如果你想讓你的資料處理具有可靠性，正确的方式是儲存這些tuple對象的引用直到bolt批量處理這些tuple了。一旦這個批量操作結束， 你可以批量的ack這些tuple。

如果一個bolt發射tuple， 那麼你可能想用multi-anchoring來保證可靠性。這一切都取決于具體的應用。關于storm的消息傳遞的工作原理可以看這篇: Twitter Storm如何保證消息不丢失。

BasicBolt

很多bolt有些類似的模式:

1. 讀一個輸入tuple
2. 根據這個輸入tuple發射一個或者多個tuple
3. 在execute的方法的最後ack那個輸入tuple

遵循這類模式的bolt一般是函數或者是過濾器, 這種模式太常見，storm為這類模式單獨封裝了一個接口: IBasicBolt。更多的資訊請看: Twitter Storm如何保證消息不丢失。

記憶體内緩存 + fields grouping 組合

在bolt的記憶體裡面緩存一些東西非常常見。緩存在和fields grouping結合起來之後就更有用了。比如，你有一個bolt把短連結變成長連結(bit.ly, t.co之類的)。你可以把短連結到長連結的對應關系利用LRU算法緩存在記憶體裡面以避免重複計算。比如元件一發射短連結，元件二把短連結轉化成長連結并緩存在記憶體裡面。看一下下面兩段代碼有什麼不一樣：

幫助

1 2	builder.setBolt( 2 , new ExpandUrl(), parallelism) .shuffleGrouping( 1 );

幫助

1 2	builder.setBolt( 2 , new ExpandUrl(), parallelism) .fieldsGrouping( 1 , new Fields( "url" ));

第二種方式的緩存會比第一種方式的緩存的效率高很多，因為同樣的短連結始終被發到同一個task。這會避免不同的機器上有同樣的緩存  — 浪費記憶體， 同時也使得同樣的短域名更可能在記憶體裡面找到緩存。

計算top N

storm的一個常見的持續計算的模式叫做: “streaming top N”。

比如你有一個bolt發射這樣的tuple: ["value", "count"]并且你想一個bolt基于這些資訊算出top N的tuple。最簡單的辦法是有一個bolt可以做一個全局的grouping的動作并且在記憶體裡面保持這top N的值。

這個方式對于大資料量的流顯然是沒有擴充性的， 因為所有的資料會被發到同一台機器， 單機的處理能力始終是有極限的。一個更好的方法是在多台機器上面并行的計算這個流每一部分的top N, 然後再有一個bolt合并這些機器上面所算出來的top N以算出最後的top N(Map Reduce的思想), 代碼大概是這樣的:

幫助

1 2 3 4

builder.setBolt( 2 , new RankObjects(), parallellism) .fieldsGrouping( 1 , new Fields( "value" )); builder.setBolt( 3 , new MergeObjects()) .globalGrouping( 2 );

這個模式之是以可行是因為第一個bolt的fields grouping使得這種并行算法在語義上是正确的。

用TimeCacheMap來高效地儲存一個最近被更新的對象的緩存

有時候你想在記憶體裡面儲存一些最近活躍的對象，以及讓那些不再活躍的對象自動過期（删除掉）。TimeCacheMap是一個非常高效的資料結構，它提供了一些callback函數使得我們在對象不再活躍的時候做一些事情。關于TimeCacheMap為什麼高效，可以看看這篇分析文章

分布式RPC: CoordinatedBolt和KeyedFairBolt

用storm做分布式RPC應用的時候有兩種比較常見的模式：它們被封裝在 CoordinatedBolt和KeyedFairBolt裡面。

CoordinatedBolt包裝你的bolt，并且确定什麼時候你的bolt已經接收到所有的tuple。它主要使用Direct Stream來做這個。

KeyedFairBolt同樣包裝你的bolt并且保證你的topology同時處理多個DRPC調用，而不是串行地一次隻執行一個。

更多有關分布式RPC的資訊可以看這裡。