Storm Trident示例Aggregator
Aggregator首先在輸入流上運行全局重新分區操作(global)將同一批次的所有分區合並到一個分區中,然後在每個批次上運行的聚合功能,針對Batch操作。與ReduceAggregator很相似。
省略部分代碼,省略部分可參考:https://blog.csdn.net/nickta/article/details/79666918
static class State { int count = 0; }
FixedBatchSpout spout = new FixedBatchSpout(new Fields("user", "score"), 3,new Values("nickt1", 4), new Values("nickt2", 7), new Values("nickt3", 8), new Values("nickt4", 9), new Values("nickt5", 7), new Values("nickt6", 11), new Values("nickt7", 5) ); spout.setCycle(false); TridentTopology topology = new TridentTopology(); topology.newStream("spout1", spout) .shuffle() .each(new Fields("user", "score"),new Debug("shuffle print:")) .parallelismHint(5) .aggregate(new Fields("score"), newBaseAggregator<State>() { //在處理每一個batch的數據之前,調用1次 //空batch也會調用 @Override public State init(Object batchId, TridentCollector collector) { return new State(); } //batch中的每個tuple各調用1次 @Override public void aggregate(State state, TridentTuple tuple, TridentCollector collector) { state.count = tuple.getInteger(0) + state.count; } //batch中的所有tuples處理完成後調用 @Override public void complete(State state, TridentCollector collector) { collector.emit(new Values(state.count)); } }, new Fields("sum")) .each(new Fields("sum"),new Debug("sum print:")) .parallelismHint(5);
輸出:
[partition4-Thread-136-b-0-executor[37 37]]> DEBUG(shuffle print:): [nickt1, 4]
[partition4-Thread-136-b-0-executor[37 37]]> DEBUG(shuffle print:): [nickt3, 8]
[partition3-Thread-118-b-0-executor[36 36]]> DEBUG(shuffle print:): [nickt2, 7]
[partition4-Thread-136-b-0-executor[37 37]]> DEBUG(shuffle print:): [nickt5, 7]
[partition3-Thread-118-b-0-executor[36 36]]> DEBUG(shuffle print:): [nickt4, 9]
[partition3-Thread-118-b-0-executor[36 36]]> DEBUG(shuffle print:): [nickt6, 11]
[partition1-Thread-82-b-1-executor[39 39]]> DEBUG(sum print:): [19]
[partition2-Thread-66-b-1-executor[40 40]]> DEBUG(sum print:): [27]
[partition4-Thread-136-b-0-executor[37 37]]> DEBUG(shuffle print:): [nickt7, 5]
[partition3-Thread-54-b-1-executor[41 41]]> DEBUG(sum print:): [5]
Storm Trident示例Aggregator