1. 程式人生 > >Storm Trident示例function, filter, projection

Storm Trident示例function, filter, projection

部分 tin keep class top collect storm topo .get

以下代碼演示function, filter, projection的使用,可結合註釋

省略部分代碼,省略部分可參考:https://blog.csdn.net/nickta/article/details/79666918

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"),new BaseFilter() { @Override public boolean isKeep(TridentTuple tuple) { if(tuple.getString(0).equals("nickt2")) { return false; } return true
; } })//過濾點user為nickt2的tuple .each(new Fields("user", "score"),new Debug("filter print:")) .each(new Fields("score"), new BaseFunction() { @Override public void execute(TridentTuple tuple, TridentCollector collector) { collector.emit(new Values(tuple.getIntegerByField("score") + 100)); } }, new Fields("sum"))//把score加上100後,生成新的sum字段,並追加到原字段後面,此步操作後就包括了user/score/sum三個字段 .each(new Fields("user", "score", "sum"),new Debug("function print:")) .project(new Fields("user")) .each(new Fields("user"),new Debug("project print:"));//project投射之後,只有user字段了

輸出:

<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(filter print:): [nickt1, 4]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(function print:): [nickt1, 4, 104]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(project print:): [nickt1]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(filter print:): [nickt3, 8]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(function print:): [nickt3, 8, 108]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(project print:): [nickt3]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(filter print:): [nickt4, 9]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(function print:): [nickt4, 9, 109]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(project print:): [nickt4]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(filter print:): [nickt5, 7]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(function print:): [nickt5, 7, 107]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(project print:): [nickt5]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(filter print:): [nickt6, 11]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(function print:): [nickt6, 11, 111]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(project print:): [nickt6]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(filter print:): [nickt7, 5]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(function print:): [nickt7, 5, 105]
<Sat Mar 24 13:41:42 CST 2018[partition0-Thread-68-b-0-executor[33 33]]> DEBUG(project print:): [nickt7]

Storm Trident示例function, filter, projection