783      技術社區[雲棲]

Storm專題二：Storm Trident API 使用詳解

     

    public class MyFunction extends BaseFunction {
      @Override
      public void execute(TridentTuple tuple, TridentCollector collector) {
           for ( int i = 0; i < tuple.getInteger(0); i++) {
              collector.emit( new Values(i));
          }
     }
   }

 java mystream.each(new Fields("b"), new MyFunction(), new Fields("d")))；

 public class MyFilter extends BaseFilter {
     public boolean isKeep(TridentTuple tuple) {
           return tuple.getInteger(0) == 1 && tuple.getInteger(1) == 2;
     }
   }

 java mystream.each(new Fields("b", "a"), new MyFilter())；

java mystream.partitionAggregate(new Fields("b"), new Sum(), new Fields("sum"))；

     Partition 1: [“a”, 3] [“c”, 8]

     Partition 1: [11]

 public interface CombinerAggregator <T> extends Serializable {
         T init(TridentTuple tuple);
         T combine(T val1, T val2);
         T zero();
    }

 public class CombinerCount implements CombinerAggregator<Integer>{
     @Override
     public Integer init(TridentTuple tuple) {
           return 1;
     }
     @Override
     public Integer combine(Integer val1, Integer val2) {
          
           return val1 + val2;
     }
     @Override
     public Integer zero() {
           return 0;
     }
   }

 public interface ReducerAggregator <T> extends Serializable {
         T init();
         T reduce(T curr, TridentTuple tuple);
     }

 public class ReducerCount implements ReducerAggregator<Long>{
     @Override
     public Long init() {
           return 0L;
     }
     @Override
     public Long reduce(Long curr, TridentTuple tuple) {
           return curr + 1;
     }
 }

最後一個是Aggregator接口，它是最通用的聚合器，它的形式如下：

  public interface Aggregator<T> extends Operation {
        T init(Object batchId, TridentCollector collector);
        void aggregate(T val, TridentTuple tuple, TridentCollector collector);
        void complete(T val, TridentCollector collector);
   }

    下麵舉例說明：

• 定義一個實現來完成一個計數器：

   public class CountAgg extends BaseAggregator<CountState>{
     static class CountState { long count = 0; }
     @Override
     public CountState init(Object batchId, TridentCollector collector) {
           return new CountState();
     }
     @Override
     public void aggregate(CountState val, TridentTuple tuple, TridentCollector collector) {
          val. count+=1;
     }
     @Override
     public void complete(CountState val, TridentCollector collector) {
          collector.emit( new Values(val. count));
     }
  }

    有時候我們需要同時執行多個聚合器，這在Trident中被稱作chaining,使用方法如下：

java mystream.chainedAgg() .partitionAggregate(new Count(), new Fields("count")) .partitionAggregate(new Fields("b"), new Sum(), new Fields("sum")) .chainEnd();

    這點代碼會在每個Partition上運行count和sum函數，最終輸出一個tuple：[“count”, “sum”]

projection:投影操作

     投影操作作用是僅保留Stream指定字段的數據，比如有一個Stream包含如下字段： [“a”, “b”, “c”, “d”]

     運行如下代碼：   

java mystream.project(new Fields("b", "d"))

    則輸出的流僅包含 [“b”, “d”]字段。

2.2 Repartitioning重定向操作

     重定向操作是如何在各個任務間對tuples進行分區。分區的數量也有可能改變重定向的結果。重定向需要網絡傳輸，下麵介紹下重定向函數：

1. shuffle：通過隨機分配算法來均衡tuple到各個分區
2. broadcast：每個tuple都被廣播到所有的分區，這種方式在drcp時非常有用，比如在每個分區上做stateQuery
3. partitionBy：根據指定的字段列表進行劃分，具體做法是用指定字段列表的hash值對分區個數做取模運算，確保相同字段列表的數據被劃分到同一個分區
4. global：所有的tuple都被發送到一個分區，這個分區用來處理整個Stream
5. batchGlobal：一個Batch中的所有tuple都被發送到同一個分區，不同的Batch會去往不同的分區
6. Partition：通過一個自定義的分區函數來進行分區，這個自定義函數實現了 backtype.storm.grouping.CustomStreamGrouping

     Trident有aggregate和 persistentAggregate方法來做聚合操作。aggregate是獨立的運行在Stream的每個Batch上的，而persistentAggregate則是運行在Stream的所有Batch上並把運算結果存儲在state source中。
     運行aggregate方法做全局聚合。當你用到  ReducerAggregator或Aggregator時，Stream首先被重定向到一個分區中，然後其中的聚合函數便在這個分區上運行。當你用到CombinerAggregator時，Trident會首先在每個分區上做局部聚合，然後把局部聚合後的結果重定向到一個分區，因此使用CombinerAggregator會更高效，可能的話我們需要優先考慮使用它。
     下麵舉個例子來說明如何用aggregate進行全局計數：

java mystream.aggregate(new Count(), new Fields("count"))；

2.4 grouped streams

      GroupBy操作是根據特定的字段對流進行重定向的，還有，在一個分區內部，每個相同字段的tuple也會被Group到一起，下麵這幅圖描述了這個場景：

     如果你在grouped Stream上麵運行aggregators，聚合操作會運行在每個Group中而不是整個Batch。persistentAggregate也能運行在GroupedSteam上，不過結果會被保存在MapState中，其中的key便是分組的字段。
     當然，aggregators在GroupedStreams上也可以串聯。

2.5 Merge和Joins:

api的最後一部分便是如何把各種流匯聚到一起。最簡單的方式就是把這些流匯聚成一個流。我們可以這麼做：   

java topology.merge(stream1, stream2, stream3);

 1. join的字段，如Stream1中的key和Stream2中的x    

 2. 所有非join的字段，根據傳入join方法的順序，a和b分別代表steam1的val1和val2，c代表Stream2的val1          

     當join的是來源於不同Spout的stream時，這些Spout在發射數據時需要同步，一個Batch所包含的tuple會來自各個Spout。    

關於Trident State 相關內容請參考【Trident State詳解】 

關於TridentAPI實踐方麵，請參考【Trident API實踐】

最後更新：2017-04-03 05:40:01

  上一篇： 我們的首要之務，並不是遙望模煳的遠方，而是專心處理眼前的事務

  下一篇： 程序員生存定律-打造屬於自己的稀缺性