// Path: src/main/java/tutorial/storm/trident/operations/DivideAsDouble.java
// public class DivideAsDouble extends BaseFunction {
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         Number n1 = (Number)tuple.get(0);
//         Number n2 = (Number)tuple.get(1);
//         collector.emit(new Values(n1.doubleValue() / n2.doubleValue()));
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/Print.java
// public class Print extends BaseFilter {
//     private int partitionIndex;
//     private int numPartitions;
//     private final String name;
// 
//     public Print(){
//         name = "";
//     }
//     public Print(String name){
//         this.name = name;
//     }
// 
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//         this.partitionIndex = context.getPartitionIndex();
//         this.numPartitions = context.numPartitions();
//     }
// 
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         System.err.println(String.format("%s::Partition idx: %s out of %s partitions got %s", name, partitionIndex, numPartitions, tuple.toString()));
//         return true;
//     }
// }


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import com.google.common.collect.ImmutableList;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.Sum;
import storm.trident.testing.FeederBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.DivideAsDouble;
import tutorial.storm.trident.operations.Print;
import java.io.IOException;
package tutorial.storm.trident;




/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class Part03_AdvancedPrimitives2 {

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
//        conf.put(Config.TOPOLOGY_DEBUG,true);
        LocalCluster cluster = new LocalCluster();

        // This time we use a "FeederBatchSpout", a spout designed for testing.
        FeederBatchSpout testSpout = new FeederBatchSpout(ImmutableList.of("name", "city", "age"));
        cluster.submitTopology("advanced_primitives", conf, advancedPrimitives(testSpout));

        // You can "hand feed" values to the topology by using this spout
        testSpout.feed(ImmutableList.of(new Values("rose", "Shanghai", 32), new Values("mary", "Shanghai", 51), new Values("pere", "Jakarta", 65), new Values("Tom", "Jakarta", 10)));
    }

    private static StormTopology advancedPrimitives(FeederBatchSpout spout) throws IOException {

        TridentTopology topology = new TridentTopology();

        // What if we want more than one aggregation? For that, we can use "chained" aggregations.
        // Note how we calculate count and sum.
        // The aggregated values can then be processed further, in this case into mean
        topology
                .newStream("aggregation", spout)
                .groupBy(new Fields("city"))
                .chainedAgg()
                .aggregate(new Count(), new Fields("count"))
                .aggregate(new Fields("age"), new Sum(), new Fields("age_sum"))
                .chainEnd()
                .each(new Fields("age_sum", "count"), new DivideAsDouble(), new Fields("mean_age"))
// Path: src/main/java/tutorial/storm/trident/operations/DivideAsDouble.java
// public class DivideAsDouble extends BaseFunction {
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector) {
//         Number n1 = (Number)tuple.get(0);
//         Number n2 = (Number)tuple.get(1);
//         collector.emit(new Values(n1.doubleValue() / n2.doubleValue()));
//     }
// }
// 
// Path: src/main/java/tutorial/storm/trident/operations/Print.java
// public class Print extends BaseFilter {
//     private int partitionIndex;
//     private int numPartitions;
//     private final String name;
// 
//     public Print(){
//         name = "";
//     }
//     public Print(String name){
//         this.name = name;
//     }
// 
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//         this.partitionIndex = context.getPartitionIndex();
//         this.numPartitions = context.numPartitions();
//     }
// 
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         System.err.println(String.format("%s::Partition idx: %s out of %s partitions got %s", name, partitionIndex, numPartitions, tuple.toString()));
//         return true;
//     }
// }


// Path: src/main/java/tutorial/storm/trident/Part03_AdvancedPrimitives2.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import com.google.common.collect.ImmutableList;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.Sum;
import storm.trident.testing.FeederBatchSpout;
import storm.trident.testing.MemoryMapState;
import tutorial.storm.trident.operations.DivideAsDouble;
import tutorial.storm.trident.operations.Print;
import java.io.IOException;

package tutorial.storm.trident;




/**
 * @author Enno Shioji (enno.shioji@peerindex.com)
 */
public class Part03_AdvancedPrimitives2 {

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
//        conf.put(Config.TOPOLOGY_DEBUG,true);
        LocalCluster cluster = new LocalCluster();

        // This time we use a "FeederBatchSpout", a spout designed for testing.
        FeederBatchSpout testSpout = new FeederBatchSpout(ImmutableList.of("name", "city", "age"));
        cluster.submitTopology("advanced_primitives", conf, advancedPrimitives(testSpout));

        // You can "hand feed" values to the topology by using this spout
        testSpout.feed(ImmutableList.of(new Values("rose", "Shanghai", 32), new Values("mary", "Shanghai", 51), new Values("pere", "Jakarta", 65), new Values("Tom", "Jakarta", 10)));
    }

    private static StormTopology advancedPrimitives(FeederBatchSpout spout) throws IOException {

        TridentTopology topology = new TridentTopology();

        // What if we want more than one aggregation? For that, we can use "chained" aggregations.
        // Note how we calculate count and sum.
        // The aggregated values can then be processed further, in this case into mean
        topology
                .newStream("aggregation", spout)
                .groupBy(new Fields("city"))
                .chainedAgg()
                .aggregate(new Count(), new Fields("count"))
                .aggregate(new Fields("age"), new Sum(), new Fields("age_sum"))
                .chainEnd()
                .each(new Fields("age_sum", "count"), new DivideAsDouble(), new Fields("mean_age"))
                .each(new Fields("city", "mean_age"), new Print())