// Path: src/main/java/acromusashi/stream/example/ml/trident/ResultPrintFunction.java
// public class ResultPrintFunction extends BaseFunction
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = -6332711183777478101L;
// 
//     /** Logger */
//     private static final Logger logger           = LoggerFactory.getLogger(ResultPrintFunction.class);
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public ResultPrintFunction()
//     {}
// 
//     /**
//      * {@inheritDoc}
//      */
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void prepare(Map conf, TridentOperationContext context)
//     {
//         // Do nothing.
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector)
//     {
//         ApacheLog statistics = (ApacheLog) tuple.getValueByField("average");
//         String logFormat = "Statistics Result: Host={0}, DataCount={1}, MaxChangeFindScore={2}, ResponseTimeSum={3}, SizeSum={4}";
//         logger.info(MessageFormat.format(logFormat, statistics.getKey(), statistics.getCount(),
//                 statistics.getAnomalyScore(), statistics.getTimeSum(), statistics.getSizeSum()));
//     }
// }


import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.OpaqueTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import storm.trident.operation.impl.CombinerAggregatorCombineImpl;
import acromusashi.stream.config.StormConfigGenerator;
import acromusashi.stream.config.StormConfigUtil;
import acromusashi.stream.example.ml.trident.ResultPrintFunction;
import acromusashi.stream.ml.loganalyze.ApacheLog;
import acromusashi.stream.ml.loganalyze.ApacheLogAggregator;
import acromusashi.stream.ml.loganalyze.ApacheLogSplitFunction;
import acromusashi.stream.ml.loganalyze.ChangeFindFunction;
import acromusashi.stream.topology.BaseTridentTopology;
import backtype.storm.Config;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;
        cfFunction.setForgetability(forgetability);
        cfFunction.setSmoothingWindow(smoothingWindow);
        cfFunction.setScoreThreshold(scoreThreshold);

        TridentTopology topology = new TridentTopology();

        // 以下の順でTridentTopologyにSpout/Functionを登録する。
        // 1.TridentKafkaSpout:KafkaからApacheログ(JSON形式)を取得
        //
        // ※全体並列度を「topology.parallelismHint」の設定値に併せて設定
        // 
        // >(each:各メッセージに対して実行)
        // 2.ApacheLogSplitFunction:受信したApacheログ(JSON形式)をエンティティに変換し、送信
        //
        // ※エンティティを「IPaddress」の値毎にグルーピング
        //
        // >(each:各メッセージに対して実行)
        // 3.ChangeFindFunction:受信したApacheログのレスポンスタイムを用いて変化点スコアを算出
        // 
        // >(partitionAggregate:グルーピングしたエンティティを統合)
        // 4.ApacheLogAggregator:受信したApacheログの統計値を算出
        //
        // >(each:各メッセージに対して実行)
        // 5.ResultPrintFunction:受信した統計値をログ出力
        // @formatter:off
        topology.newStream("TridentKafkaSpout", tridentKafkaSpout).parallelismHint(parallelism)
            .each(new Fields("str"), apacheLogSplitFunction, new Fields("IPaddress", "responseTime"))
            .groupBy(new Fields("IPaddress"))
            .each(new Fields("IPaddress", "responseTime"), cfFunction, new Fields("webResponse"))
            .partitionAggregate(new Fields("webResponse"), new CombinerAggregatorCombineImpl(new ApacheLogAggregator()), new Fields("average"))
// Path: src/main/java/acromusashi/stream/example/ml/trident/ResultPrintFunction.java
// public class ResultPrintFunction extends BaseFunction
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = -6332711183777478101L;
// 
//     /** Logger */
//     private static final Logger logger           = LoggerFactory.getLogger(ResultPrintFunction.class);
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public ResultPrintFunction()
//     {}
// 
//     /**
//      * {@inheritDoc}
//      */
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void prepare(Map conf, TridentOperationContext context)
//     {
//         // Do nothing.
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void execute(TridentTuple tuple, TridentCollector collector)
//     {
//         ApacheLog statistics = (ApacheLog) tuple.getValueByField("average");
//         String logFormat = "Statistics Result: Host={0}, DataCount={1}, MaxChangeFindScore={2}, ResponseTimeSum={3}, SizeSum={4}";
//         logger.info(MessageFormat.format(logFormat, statistics.getKey(), statistics.getCount(),
//                 statistics.getAnomalyScore(), statistics.getTimeSum(), statistics.getSizeSum()));
//     }
// }


// Path: src/main/java/acromusashi/stream/example/ml/topology/ChangeFindTopology.java
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.OpaqueTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.TridentTopology;
import storm.trident.operation.impl.CombinerAggregatorCombineImpl;
import acromusashi.stream.config.StormConfigGenerator;
import acromusashi.stream.config.StormConfigUtil;
import acromusashi.stream.example.ml.trident.ResultPrintFunction;
import acromusashi.stream.ml.loganalyze.ApacheLog;
import acromusashi.stream.ml.loganalyze.ApacheLogAggregator;
import acromusashi.stream.ml.loganalyze.ApacheLogSplitFunction;
import acromusashi.stream.ml.loganalyze.ChangeFindFunction;
import acromusashi.stream.topology.BaseTridentTopology;
import backtype.storm.Config;
import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.tuple.Fields;

        cfFunction.setForgetability(forgetability);
        cfFunction.setSmoothingWindow(smoothingWindow);
        cfFunction.setScoreThreshold(scoreThreshold);

        TridentTopology topology = new TridentTopology();

        // 以下の順でTridentTopologyにSpout/Functionを登録する。
        // 1.TridentKafkaSpout:KafkaからApacheログ(JSON形式)を取得
        //
        // ※全体並列度を「topology.parallelismHint」の設定値に併せて設定
        // 
        // >(each:各メッセージに対して実行)
        // 2.ApacheLogSplitFunction:受信したApacheログ(JSON形式)をエンティティに変換し、送信
        //
        // ※エンティティを「IPaddress」の値毎にグルーピング
        //
        // >(each:各メッセージに対して実行)
        // 3.ChangeFindFunction:受信したApacheログのレスポンスタイムを用いて変化点スコアを算出
        // 
        // >(partitionAggregate:グルーピングしたエンティティを統合)
        // 4.ApacheLogAggregator:受信したApacheログの統計値を算出
        //
        // >(each:各メッセージに対して実行)
        // 5.ResultPrintFunction:受信した統計値をログ出力
        // @formatter:off
        topology.newStream("TridentKafkaSpout", tridentKafkaSpout).parallelismHint(parallelism)
            .each(new Fields("str"), apacheLogSplitFunction, new Fields("IPaddress", "responseTime"))
            .groupBy(new Fields("IPaddress"))
            .each(new Fields("IPaddress", "responseTime"), cfFunction, new Fields("webResponse"))
            .partitionAggregate(new Fields("webResponse"), new CombinerAggregatorCombineImpl(new ApacheLogAggregator()), new Fields("average"))
            .each(new Fields("average"), new ResultPrintFunction(), new Fields("count"));