Storm Trident示例ReducerAggregator

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"), new ReducerAggregator<Integer>() {  
                    @Override  
                    //每一个batch，初始调用 1次，空的batch也会调用1次  
                    public Integer init() {  
                        return 0;  
                    }  
                    @Override  
                    //batch看的每个tuple调用1次  
                    public Integer reduce(Integer curr, TridentTuple tuple) {  
                        return curr + tuple.getIntegerByField("score");  
                    }  
                }, new Fields("sum"))//对每个batch中的score求和  
                .each(new Fields("sum"),new Debug("sum print:"))  
                .parallelismHint(5);