diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseFourPartATridentTopology.java b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseFourPartATridentTopology.java new file mode 100644 index 0000000..24a757a --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseFourPartATridentTopology.java @@ -0,0 +1,150 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import java.text.SimpleDateFormat; +import java.util.Date; + +import storm.trident.TridentState; +import storm.trident.TridentTopology; +import storm.trident.operation.BaseFunction; +import storm.trident.operation.TridentCollector; +import storm.trident.operation.builtin.Debug; +import storm.trident.operation.builtin.FilterNull; +import storm.trident.operation.builtin.MapGet; +import storm.trident.operation.builtin.Sum; +import storm.trident.state.StateFactory; +import storm.trident.testing.MemoryMapState; +import storm.trident.testing.Split; +import storm.trident.tuple.TridentTuple; +import backtype.storm.Config; +import backtype.storm.LocalCluster; +import backtype.storm.LocalDRPC; +import backtype.storm.generated.StormTopology; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.Values; + +public class ExerciseFourPartATridentTopology { + + private static final int DRPC_QUERY_ITERATIONS = 2000; + private static final int DRPC_QUERY_INTERVAL = 200; + // Data path relative to pom.xml file. + private static final String DATA_PATH = "data/20130301.csv.gz"; + + /** + * Launch a topology that reads stock symbols and prices from a CSV data file. Query the topology with DRPC every + * + * @param args + * @throws Exception + */ + public static void main( String[] args ) throws Exception { + Config conf = new Config(); + // conf.setDebug( true ); + conf.setMaxSpoutPending( 20 ); + + // This topology can only be run as local because it is a toy example + LocalDRPC drpc = new LocalDRPC(); + LocalCluster cluster = new LocalCluster(); + cluster.submitTopology( "symbolCounter", conf, buildTopology( drpc ) ); + + for (int i = 0; i < DRPC_QUERY_ITERATIONS; i++) { + + System.err.println( "Result for DRPC stock volume query -> " + drpc.execute( "trades", "INTC|2013-03-01-09:30:00 INTC|2013-03-01-09:31:00 INTC|2013-03-01-09:32:00" ) ); + Thread.sleep( DRPC_QUERY_INTERVAL ); + } + + } + + public static StormTopology buildTopology( LocalDRPC drpc ) { + + TridentTopology topology = new TridentTopology(); + + // Create spout to read data from CSV file and emit fields "date", "symbol", "price", "shares" + // The Fields instance is used to name the fields the parsed output fields are mapped to. + CSVBatchSpout spout = new CSVBatchSpout( DATA_PATH, new Fields( "date", "symbol", "price", "shares" ) ); + + // + // In this state we will save the real-time counts for each symbol + // For this demo, we will use an in-process memory map. We could just as easily use Memcached or a NoSQL data store + StateFactory mapState = new MemoryMapState.Factory(); + + // Real-time part of the system: a Trident topology that groups by symbol and stores per-symbol counts + TridentState tradeVolume = topology.newStream( "quotes", spout ) + // + + // Debug output of "quotes" spout + //.each( new Fields( "date", "symbol", "price", "shares" ), new Debug() ) + + // hour stamp added to key + .each( new Fields( "symbol", "date"), new SymbolMinuteKey(), new Fields( "symbol_minute" )) + + //uncomment to debug + //.each( new Fields( "symbol", "date", "symbol_minute"), new Debug()) + + // -- fields grouping by "symbol" and "hour" + .groupBy( new Fields( "symbol_minute" ) ) + + // + // Aggregate shares by symbol and hour, projecting new field "volume" + .persistentAggregate( mapState, new Fields( "shares" ), new Sum(), new Fields( "volume" ) ); + + /** + * Now setup a DRPC stream on top of the "quotes" stream. The DRPC stream will generate a list of symbols and then query the + * persistent aggregate state by symbol for the volume value. + */ + topology.newDRPCStream( "trades", drpc ) + // + + // Freaking awesome DEBUG!!! + // This debug statement will emit stock symbols: DEBUG: [INTC GE AAPL] + // .each( new Fields( "args" ), new Debug() ) + + // state query. The input is always "args", and needs to be split into individual fields + // Split() implements tuple.getString(0).split(" ") + .each( new Fields( "args" ), new Split(), new Fields( "symbol_minute" ) ) + // + // .each( new Fields( "symbol_minute" ), new Debug() ) + // + .groupBy( new Fields( "symbol_minute" ) ) + + // + // Query that persistent state that we setup earlier + // Query key value field is "symbol" and result is projected as "volume" + .stateQuery( tradeVolume, new Fields( "symbol_minute" ), new MapGet(), new Fields( "volume" ) ) + + // remove nulls for 'each' value in the stream (aka Filtering) + .each( new Fields( "volume" ), new FilterNull() ) + + // + // Debug print symbol and volume values + // .each( new Fields( "symbol", "volume" ), new Debug() ) + + // Remove this line to revert back to per stock symbol aggregate + //.aggregate(new Fields("volume"), new Sum(), new Fields ("total_volume")) + + // + // Project allows us to keep only the interesting fields that interest us + //.project( new Fields( "symbol", "volume") ); + + // Project total volume of trades for all stocks not just individuals + //.project( new Fields("total_volume")) + ; + + return topology.build(); + } + + private static final SimpleDateFormat FORMATTER = new SimpleDateFormat( "yyyy-MM-dd-HH:mm:ss" ); + + public static class SymbolMinuteKey extends BaseFunction { + + @Override + public void execute(TridentTuple tuple, TridentCollector collector) { + Date date = (Date)tuple.getValueByField("date"); + date = new Date(date.getTime()); // new object to not mutate original tuple! + //alternatively, just format with :00 instead of :ss + date.setSeconds(0); + String compoundKey = tuple.getStringByField("symbol")+"|"+FORMATTER.format(date); + collector.emit(new Values(compoundKey)); + } + + } +} diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseFourPartBTridentTopology.java b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseFourPartBTridentTopology.java new file mode 100644 index 0000000..6094e0c --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseFourPartBTridentTopology.java @@ -0,0 +1,191 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import storm.trident.Stream; +import storm.trident.TridentState; +import storm.trident.TridentTopology; +import storm.trident.operation.BaseFilter; +import storm.trident.operation.BaseFunction; +import storm.trident.operation.TridentCollector; +import storm.trident.operation.builtin.FilterNull; +import storm.trident.operation.builtin.MapGet; +import storm.trident.operation.builtin.Sum; +import storm.trident.state.StateFactory; +import storm.trident.testing.MemoryMapState; +import storm.trident.testing.Split; +import storm.trident.tuple.TridentTuple; +import backtype.storm.Config; +import backtype.storm.LocalCluster; +import backtype.storm.LocalDRPC; +import backtype.storm.generated.StormTopology; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.Values; + +public class ExerciseFourPartBTridentTopology { + + private static final int DRPC_QUERY_ITERATIONS = 2000; + private static final int DRPC_QUERY_INTERVAL = 200; + // Data path relative to pom.xml file. + private static final String DATA_PATH = "data/20130301.csv.gz"; + + /** + * Launch a topology that reads stock symbols and prices from a CSV data file. Query the topology with DRPC every + * + * @param args + * @throws Exception + */ + public static void main( String[] args ) throws Exception { + Config conf = new Config(); + // conf.setDebug( true ); + conf.setMaxSpoutPending( 20 ); + + // This topology can only be run as local because it is a toy example + LocalDRPC drpc = new LocalDRPC(); + LocalCluster cluster = new LocalCluster(); + cluster.submitTopology( "symbolCounter", conf, buildTopology( drpc ) ); + + + for (int i = 0; i < DRPC_QUERY_ITERATIONS; i++) { + List val = seen.get("INTC"); + if (val != null) { + String last = val.get(val.size()-1); + System.err.println("Last minute seen for INTC: "+last); + System.err.println( "Result for DRPC stock volume query -> " + drpc.execute( "trades", last ) ); + } + Thread.sleep( DRPC_QUERY_INTERVAL ); + } + + } + + // hash map to simulate a NoSQL database + private static final Map> seen = new HashMap>(); + + public static class StoreSymbolMinute extends BaseFilter { + + @Override + public boolean isKeep(TridentTuple tuple) { + String symbolMinute = tuple.getStringByField("symbol_minute"); + String symbol = symbolMinute.split("\\|")[0]; + synchronized(seen) { + List minutes = seen.get(symbol); + if (minutes == null) { + minutes = new ArrayList(); + seen.put(symbol, minutes); + } + if (!minutes.contains(symbolMinute)) { + minutes.add(symbolMinute); + } + } + return false; + } + + } + + public static StormTopology buildTopology( LocalDRPC drpc ) { + + TridentTopology topology = new TridentTopology(); + + // Create spout to read data from CSV file and emit fields "date", "symbol", "price", "shares" + // The Fields instance is used to name the fields the parsed output fields are mapped to. + CSVBatchSpout spout = new CSVBatchSpout( DATA_PATH, new Fields( "date", "symbol", "price", "shares" ) ); + + // + // In this state we will save the real-time counts for each symbol + // For this demo, we will use an in-process memory map. We could just as easily use Memcached or a NoSQL data store + StateFactory mapState = new MemoryMapState.Factory(); + + Stream stream = topology.newStream( "quotes", spout ); + + // Real-time part of the system: a Trident topology that groups by symbol and stores per-symbol counts + Stream streamWithSymbolMinute = stream + // + + // Debug output of "quotes" spout + //.each( new Fields( "date", "symbol", "price", "shares" ), new Debug() ) + + // hour stamp added to key + .each( new Fields( "symbol", "date"), new SymbolMinuteKey(), new Fields( "symbol_minute" )); + + TridentState tradeVolume = streamWithSymbolMinute + + //uncomment to debug + //.each( new Fields( "symbol", "date", "symbol_minute"), new Debug()) + + // -- fields grouping by "symbol" and "hour" + .groupBy( new Fields( "symbol_minute" ) ) + + // + // Aggregate shares by symbol and hour, projecting new field "volume" + .persistentAggregate( mapState, new Fields( "shares" ), new Sum(), new Fields( "volume" ) ); + + streamWithSymbolMinute. + each(new Fields("symbol_minute"), new StoreSymbolMinute()); + + /** + * Now setup a DRPC stream on top of the "quotes" stream. The DRPC stream will generate a list of symbols and then query the + * persistent aggregate state by symbol for the volume value. + */ + topology.newDRPCStream( "trades", drpc ) + // + + // Freaking awesome DEBUG!!! + // This debug statement will emit stock symbols: DEBUG: [INTC GE AAPL] + // .each( new Fields( "args" ), new Debug() ) + + // state query. The input is always "args", and needs to be split into individual fields + // Split() implements tuple.getString(0).split(" ") + .each( new Fields( "args" ), new Split(), new Fields( "symbol_minute" ) ) + // + // .each( new Fields( "symbol_minute" ), new Debug() ) + // + .groupBy( new Fields( "symbol_minute" ) ) + + // + // Query that persistent state that we setup earlier + // Query key value field is "symbol" and result is projected as "volume" + .stateQuery( tradeVolume, new Fields( "symbol_minute" ), new MapGet(), new Fields( "volume" ) ) + + // remove nulls for 'each' value in the stream (aka Filtering) + .each( new Fields( "volume" ), new FilterNull() ) + + // + // Debug print symbol and volume values + // .each( new Fields( "symbol", "volume" ), new Debug() ) + + // Remove this line to revert back to per stock symbol aggregate + //.aggregate(new Fields("volume"), new Sum(), new Fields ("total_volume")) + + // + // Project allows us to keep only the interesting fields that interest us + //.project( new Fields( "symbol", "volume") ); + + // Project total volume of trades for all stocks not just individuals + //.project( new Fields("total_volume")) + ; + + return topology.build(); + } + + private static final SimpleDateFormat FORMATTER = new SimpleDateFormat( "yyyy-MM-dd-HH:mm:ss" ); + + public static class SymbolMinuteKey extends BaseFunction { + + @Override + public void execute(TridentTuple tuple, TridentCollector collector) { + Date date = (Date)tuple.getValueByField("date"); + date = new Date(date.getTime()); // new object to not mutate original tuple! + //alternatively, just format with :00 instead of :ss + date.setSeconds(0); + String compoundKey = tuple.getStringByField("symbol")+"|"+FORMATTER.format(date); + collector.emit(new Values(compoundKey)); + } + + } +} diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseOneTridentTopology.java b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseOneTridentTopology.java new file mode 100644 index 0000000..28ae09d --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseOneTridentTopology.java @@ -0,0 +1,117 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import storm.trident.TridentState; +import storm.trident.TridentTopology; +import storm.trident.operation.BaseFunction; +import storm.trident.operation.TridentCollector; +import storm.trident.operation.builtin.Debug; +import storm.trident.operation.builtin.FilterNull; +import storm.trident.operation.builtin.MapGet; +import storm.trident.state.StateFactory; +import storm.trident.testing.MemoryMapState; +import storm.trident.testing.Split; +import storm.trident.tuple.TridentTuple; +import backtype.storm.Config; +import backtype.storm.LocalCluster; +import backtype.storm.LocalDRPC; +import backtype.storm.StormSubmitter; +import backtype.storm.generated.StormTopology; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.Values; + +/*- + * Demonstrate using a custom CombinerAggregator to manage multi-valued stock statistics. + */ +public class ExerciseOneTridentTopology { + + public static class RoundFunction extends BaseFunction { + + @Override + public void execute(TridentTuple tuple, TridentCollector collector) { + collector.emit(new Values(new Double(Math.round(tuple.getDoubleByField( "price" ))))); + } + + } + + public static final String DRPC_STREAM_NAME = "trades"; + private static final String STOCK_TRIDENT_TOPOLOGY_NAME = "StockTridentTopology"; + private static final String DATA_PATH = "data/20130301.csv.gz"; + + private static final Fields fields = new Fields( "date", "symbol", "price", "shares" ); + + public static void main( String[] args ) throws Exception { + Config conf = new Config(); + conf.setDebug( false ); + conf.setMaxSpoutPending( 1 ); + final boolean isLocal = true; + + if (isLocal) { + // This topology can only be run as local because it is a toy example + LocalDRPC drpc = new LocalDRPC(); + LocalCluster cluster = new LocalCluster(); + cluster.submitTopology( STOCK_TRIDENT_TOPOLOGY_NAME, conf, buildTopology( drpc ) ); + + for (int i = 0; i < 2000; i++) { + /** + * Run a query passing in symbols to be examined. + */ + System.err.println( "DRPC response: " + drpc.execute( DRPC_STREAM_NAME, "INTC GE AAPL" ) ); + Thread.sleep( 1000 ); + } + } else { + StormSubmitter.submitTopology( STOCK_TRIDENT_TOPOLOGY_NAME, conf, buildTopology( null ) ); + } + } + + public static StormTopology buildTopology( LocalDRPC drpc ) { + + TridentTopology topology = new TridentTopology(); + CSVBatchSpout spout = new CSVBatchSpout( DATA_PATH, fields ); + + // In this state we will save the real-time counts for each symbol + StateFactory mapState = new MemoryMapState.Factory(); + + // Real-time part of the system: a Trident topology that groups by symbol and stores per-symbol counts + TridentState tradeCount = topology.newStream( "quotes", spout ) + // + .each(fields, new RoundFunction(), new Fields("dollar_price")) + + // debug display fields values. + //.each( new Fields("dollar_price","price"), new Debug() ) + // + + /*- + * Use fieldsGrouping on "symbol" field. This is essentially a groupby operation. + */ + .groupBy( new Fields( "symbol" ) ) + /*- + * fields are processed by StockAggregator and combined with a new object stock "stats" emitted + * on a continuous basis. + */ + .persistentAggregate( mapState, new Fields("dollar_price","price","symbol","shares"), new StockAggregatorRounded(), new Fields( "stats" ) ); + + /*- + * Pretty much same DRPC stream as used in SimpleTridentTopology. + * We're retrieving a more complex multi-valued structured from the + * persistent aggregate compared to a simple "volume" measure. + */ + topology.newDRPCStream( DRPC_STREAM_NAME, drpc ) + // state query + .each( new Fields( "args" ), new Split(), new Fields( "symbol" ) ) + // + .groupBy( new Fields( "symbol" ) ) + // + .stateQuery( tradeCount, new Fields( "symbol" ), new MapGet(), new Fields( "stats" ) ) + // + // Remove nulls or empty stats objects. + .each( new Fields( "stats" ), new FilterNull() ) + // + //.each( new Fields( "symbol", "stats" ), new Debug() ) + // + // Project allows us to keep only the interesting results + .project( new Fields( "symbol", "stats" ) ); + + return topology.build(); + } +} diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseThreeSimpleTridentTopology.java b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseThreeSimpleTridentTopology.java new file mode 100644 index 0000000..157d5a8 --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseThreeSimpleTridentTopology.java @@ -0,0 +1,131 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import storm.trident.TridentState; +import storm.trident.TridentTopology; +import storm.trident.operation.BaseFunction; +import storm.trident.operation.TridentCollector; +import storm.trident.operation.builtin.FilterNull; +import storm.trident.operation.builtin.MapGet; +import storm.trident.operation.builtin.Sum; +import storm.trident.state.StateFactory; +import storm.trident.testing.MemoryMapState; +import storm.trident.testing.Split; +import storm.trident.tuple.TridentTuple; +import backtype.storm.Config; +import backtype.storm.LocalCluster; +import backtype.storm.LocalDRPC; +import backtype.storm.generated.StormTopology; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.Values; + +public class ExerciseThreeSimpleTridentTopology { + + public static class Constant extends BaseFunction { + + private String constant; + + public Constant(String constant) { + this.constant = constant; + } + + @Override + public void execute(TridentTuple tuple, TridentCollector collector) { + collector.emit(new Values(constant)); + } + + + } + + private static final int DRPC_QUERY_ITERATIONS = 2000; + private static final int DRPC_QUERY_INTERVAL = 200; + // Data path relative to pom.xml file. + private static final String DATA_PATH = "data/20130301.csv.gz"; + + /** + * Launch a topology that reads stock symbols and prices from a CSV data file. Query the topology with DRPC every + * + * @param args + * @throws Exception + */ + public static void main( String[] args ) throws Exception { + Config conf = new Config(); + // conf.setDebug( true ); + conf.setMaxSpoutPending( 20 ); + + // This topology can only be run as local because it is a toy example + LocalDRPC drpc = new LocalDRPC(); + LocalCluster cluster = new LocalCluster(); + cluster.submitTopology( "symbolCounter", conf, buildTopology( drpc ) ); + + for (int i = 0; i < DRPC_QUERY_ITERATIONS; i++) { + System.err.println( "Result for DRPC stock volume query -> " + drpc.execute( "trades", "INTC GE AAPL" ) ); + Thread.sleep( DRPC_QUERY_INTERVAL ); + } + + } + + public static StormTopology buildTopology( LocalDRPC drpc ) { + + TridentTopology topology = new TridentTopology(); + + // Create spout to read data from CSV file and emit fields "date", "symbol", "price", "shares" + // The Fields instance is used to name the fields the parsed output fields are mapped to. + CSVBatchSpout spout = new CSVBatchSpout( DATA_PATH, new Fields( "date", "symbol", "price", "shares" ) ); + + // + // In this state we will save the real-time counts for each symbol + // For this demo, we will use an in-process memory map. We could just as easily use Memcached or a NoSQL data store + StateFactory mapState = new MemoryMapState.Factory(); + + // Real-time part of the system: a Trident topology that groups by symbol and stores per-symbol counts + TridentState tradeVolume = topology.newStream( "quotes", spout ) + // + + // Debug output of "quotes" spout + // .each( new Fields( "date", "symbol", "price", "shares" ), new Debug() ) + .each(new Fields("symbol"), new Constant("*"), new Fields("aggregate")) + + // -- fields grouping by "symbol" + .groupBy( new Fields( "aggregate" ) ) + + // + // Aggregate shares by symbol projecting new field "volume" + .persistentAggregate( mapState, new Fields( "shares" ), new Sum(), new Fields( "volume" ) ); + + /** + * Now setup a DRPC stream on top of the "quotes" stream. The DRPC stream will generate a list of symbols and then query the + * persistent aggregate state by symbol for the volume value. + */ + topology.newDRPCStream( "trades", drpc ) + // + + // Freaking awesome DEBUG!!! + // This debug statement will emit stock symbols: DEBUG: [INTC GE AAPL] + // .each( new Fields( "args" ), new Debug() ) + + .each( new Fields( "args" ), new Constant("*"), new Fields( "aggregate" ) ) + // + // .each( new Fields( "symbol" ), new Debug() ) + // + .groupBy( new Fields( "aggregate" ) ) + + // + // Query that persistent state that we setup earlier + // Query key value field is "symbol" and result is projected as "volume" + .stateQuery( tradeVolume, new Fields( "aggregate" ), new MapGet(), new Fields( "volume" ) ) + + // remove nulls for 'each' value in the stream (aka Filtering) + .each( new Fields( "volume" ), new FilterNull() ) + + // + // Debug print symbol and volume values + // .each( new Fields( "symbol", "volume" ), new Debug() ) + + // + // Project allows us to keep only the interesting fields that interest us + .project( new Fields( "volume" ) ); + + return topology.build(); + } +} diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseTwoTridentTopology.java b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseTwoTridentTopology.java new file mode 100644 index 0000000..156598b --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/ExerciseTwoTridentTopology.java @@ -0,0 +1,126 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import storm.trident.TridentState; +import storm.trident.TridentTopology; +import storm.trident.operation.builtin.Count; +import storm.trident.operation.builtin.Debug; +import storm.trident.operation.builtin.FilterNull; +import storm.trident.operation.builtin.MapGet; +import storm.trident.operation.builtin.Sum; +import storm.trident.state.StateFactory; +import storm.trident.testing.MemoryMapState; +import storm.trident.testing.Split; +import backtype.storm.Config; +import backtype.storm.LocalCluster; +import backtype.storm.LocalDRPC; +import backtype.storm.generated.StormTopology; +import backtype.storm.tuple.Fields; + +public class ExerciseTwoTridentTopology { + + private static final int DRPC_QUERY_ITERATIONS = 2000; + private static final int DRPC_QUERY_INTERVAL = 600; + // Data path relative to pom.xml file. + private static final String DATA_PATH = "data/20130301.csv.gz"; + + /** + * Launch a topology that reads stock symbols and prices from a CSV data file. Query the topology with DRPC every + * + * @param args + * @throws Exception + */ + public static void main( String[] args ) throws Exception { + Config conf = new Config(); + // conf.setDebug( true ); + conf.setMaxSpoutPending( 20 ); + + // This topology can only be run as local because it is a toy example + LocalDRPC drpc = new LocalDRPC(); + LocalCluster cluster = new LocalCluster(); + cluster.submitTopology( "symbolCounter", conf, buildTopology( drpc ) ); + + for (int i = 0; i < DRPC_QUERY_ITERATIONS; i++) { + System.err.println( "Result for DRPC stock volume query -> " + drpc.execute( "trades", "INTC GE AAPL" ) ); + Thread.sleep( DRPC_QUERY_INTERVAL ); + } + + } + + public static StormTopology buildTopology( LocalDRPC drpc ) { + + TridentTopology topology = new TridentTopology(); + + // Create spout to read data from CSV file and emit fields "date", "symbol", "price", "shares" + // The Fields instance is used to name the fields the parsed output fields are mapped to. + CSVBatchSpout spout = new CSVBatchSpout( DATA_PATH, new Fields( "date", "symbol", "price", "shares" ) ); + + // + // In this state we will save the real-time counts for each symbol + // For this demo, we will use an in-process memory map. We could just as easily use Memcached or a NoSQL data store + StateFactory mapState = new MemoryMapState.Factory(); + + // Real-time part of the system: a Trident topology that groups by symbol and stores per-symbol counts + TridentState tradeVolume = topology.newStream( "quotes", spout ) + // + + // Debug output of "quotes" spout + // .each( new Fields( "date", "symbol", "price", "shares" ), new Debug() ) + + // .each( new Fields ("date", "symbol", "shares"), new Debug()) + + // -- fields grouping by "symbol" + .groupBy( new Fields( "symbol" ) ) + + // + // Aggregate shares by symbol projecting new field "volume" + .persistentAggregate( mapState, new Fields( "shares"), new Sum(), new Fields( "volume")); + + /** + * Now setup a DRPC stream on top of the "quotes" stream. The DRPC stream will generate a list of symbols and then query the + * persistent aggregate state by symbol for the volume value. + */ + topology.newDRPCStream( "trades", drpc ) + // + + // Freaking awesome DEBUG!!! + // This debug statement will emit stock symbols: DEBUG: [INTC GE AAPL] + // .each( new Fields( "args" ), new Debug() ) + + // state query. The input is always "args", and needs to be split into individual fields + // Split() implements tuple.getString(0).split(" ") + .each( new Fields( "args" ), new Split(), new Fields( "symbol" ) ) + // + // .each( new Fields( "symbol" ), new Debug() ) + // + .groupBy( new Fields( "symbol" ) ) + + // + // Query that persistent state that we setup earlier + // Query key value field is "symbol" and result is projected as "volume" + .stateQuery( tradeVolume, new Fields( "symbol" ), new MapGet(), new Fields( "volume" ) ) + + // remove nulls for 'each' value in the stream (aka Filtering) + .each( new Fields( "volume" ), new FilterNull() ) + + // + // Debug print symbol and volume values + // .each( new Fields( "symbol", "volume" ), new Debug() ) + + .each( new Fields( "volume"), new StockVolumeFilter( 1000000 ) ) + // Remove this line to revert back to per stock symbol aggregate + // .aggregate(new Fields("volume"), new Sum(), new Fields ("total_volume")) + + // + // Project allows us to keep only the interesting fields that interest us + // .project( new Fields( "symbol", "volume") ); + + // Project total volume of trades for all stocks not just individuals + //.project( new Fields("total_volume")); + + // Project only the stock symbol with total volume over 10k + .project( new Fields( "symbol") ); + + return topology.build(); + } +} diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/StockAggregatorRounded.java b/answers/jvm/com/thinkbiganalytics/storm/trident/StockAggregatorRounded.java new file mode 100644 index 0000000..fff18ff --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/StockAggregatorRounded.java @@ -0,0 +1,58 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import java.util.HashMap; +import java.util.Map; + +import com.thinkbiganalytics.storm.pojo.StockStats; + +import storm.trident.operation.CombinerAggregator; +import storm.trident.tuple.TridentTuple; + +/*- + * Manage complex stock trade statistics within the Storm Trident framework. + * Inputs: symbol, shares, price + * Outputs: StockStats + */ +public class StockAggregatorRounded implements CombinerAggregator> { + + /** + * Convert the TridentTuple into a map object. + */ + @Override + public Map init( TridentTuple tuple ) { + + Map map = zero(); + final String symbol = tuple.getStringByField( "symbol" ); + final long shares = tuple.getLongByField( "shares" ); + final double dollar_price = tuple.getDoubleByField( "dollar_price" ); + assert dollar_price == new Double(Math.round(dollar_price)); + + //map.put( symbol, new StockStats( price, shares ) ); + map.put( symbol, new StockStats( dollar_price, shares ) ); + return map; + } + + /** + * Handle hash map based combine logic + */ + @Override + public Map combine( Map val1, Map val2 ) { + for (Map.Entry entry : val2.entrySet()) { + val2.put( entry.getKey(), new StockStats( val1.get( entry.getKey() ), entry.getValue() ) ); + } + for (Map.Entry entry : val1.entrySet()) { + if (val2.containsKey( entry.getKey() )) { + continue; + } + val2.put( entry.getKey(), entry.getValue() ); + } + return val2; + } + + @Override + public Map zero() { + return new HashMap(); + } + +} \ No newline at end of file diff --git a/answers/jvm/com/thinkbiganalytics/storm/trident/StockVolumeFilter.java b/answers/jvm/com/thinkbiganalytics/storm/trident/StockVolumeFilter.java new file mode 100644 index 0000000..d1d17f9 --- /dev/null +++ b/answers/jvm/com/thinkbiganalytics/storm/trident/StockVolumeFilter.java @@ -0,0 +1,28 @@ +// Copyright (c) 2013, Think Big Analytics. +package com.thinkbiganalytics.storm.trident; + +import storm.trident.operation.BaseFilter; +import storm.trident.tuple.TridentTuple; + +/*- + * Filter volume levels within the Storm Trident framework. + * Inputs: volume + * Outputs: kept tuple + */ +public class StockVolumeFilter extends BaseFilter { + + /** + * modify the filter so that only tuples with a volume greater than a certain volume are kept + */ + long volume; + + public StockVolumeFilter (long inVolume) { + this.volume = inVolume; + } + + @Override + public boolean isKeep(TridentTuple tuple) { + return tuple.getLong(0) > volume; + } + +} \ No newline at end of file diff --git a/presentations/ThinkBig_Storm_Trident_Presentation.pptx b/presentations/ThinkBig_Storm_Trident_Presentation.pptx new file mode 100644 index 0000000..1abb9ce Binary files /dev/null and b/presentations/ThinkBig_Storm_Trident_Presentation.pptx differ diff --git a/src/jvm/com/thinkbiganalytics/storm/pojo/StockStats.java b/src/jvm/com/thinkbiganalytics/storm/pojo/StockStats.java index f21d496..ca2ff23 100644 --- a/src/jvm/com/thinkbiganalytics/storm/pojo/StockStats.java +++ b/src/jvm/com/thinkbiganalytics/storm/pojo/StockStats.java @@ -23,9 +23,9 @@ public StockStats( final StockStats val1, final StockStats val2 ) { init( 0.0, 0L ); return; } else if (null == val1 && !(null == val2)) { - init( val2.avgPrice, val2.shareVolume ); + init( val2 ); } else if (!(null == val1) && (null == val2)) { - init( val1.avgPrice, val1.shareVolume ); + init( val1 ); } else { count = val1.count + val2.count; @@ -42,7 +42,18 @@ public StockStats( final StockStats val1, final StockStats val2 ) { } } - public StockStats( final double price, final long shares ) { + private void init(StockStats val) { + count = val.count; + sum = val.sum; + low = val.low; + high = val.high; + shareVolume = val.shareVolume; + priceVolume = val.priceVolume; + avgPrice = sum / count; + weightedAvgPrice = priceVolume / shareVolume; + } + + public StockStats( final double price, final long shares ) { init( price, shares ); }