Apache Storm Tutorial

1. Aggregation
   
2. Aggregation is an object used to perform aggregation operations on an input batch or partition or stream. Trident has three types of aggregation. They are as follows −
   
3. 
   
4. aggregate − Aggregates each batch of trident tuple in isolation. During the aggregate process, the tuples are initially repartitioned using the global grouping to combine all partitions of the same batch into a single partition.
   
5. 
   
6. partitionAggregate − Aggregates each partition instead of the entire batch of trident tuple. The output of the partition aggregate completely replaces the input tuple. The output of the partition aggregate contains a single field tuple.
   
7. 
   
8. persistentaggregate − Aggregates on all trident tuple across all batch and stores the result in either memory or database.
   
9. 
   
10. TridentTopology topology = new TridentTopology();
    
11. 
    
12. // aggregate operation
    
13. topology.newStream("spout", spout)
    
14.    .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
    
15.    .aggregate(new Count(), new Fields(“count”))
    
16.        
    
17. // partitionAggregate operation
    
18. topology.newStream("spout", spout)
    
19.    .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
    
20.    .partitionAggregate(new Count(), new Fields(“count"))
    
21.        
    
22. // persistentAggregate - saving the count to memory
    
23. topology.newStream("spout", spout)
    
24.    .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
    
25.    .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"));

复制代码

1. Grouping
   
2. Grouping operation is an inbuilt operation and can be called by the groupBy method. The groupBy method repartitions the stream by doing a partitionBy on the specified fields, and then within each partition, it groups tuples together whose group fields are equal. Normally, we use “groupBy” along with “persistentAggregate” to get the grouped aggregation. The sample code is as follows −
   
3. 
   
4. TridentTopology topology = new TridentTopology();
   
5. 
   
6. // persistentAggregate - saving the count to memory
   
7. topology.newStream("spout", spout)
   
8.    .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
   
9.    .groupBy(new Fields(“d”)
   
10.    .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"));

复制代码

1. Merging and Joining
   
2. Merging and joining can be done by using “merge” and “join” method respectively. Merging combines one or more streams. Joining is similar to merging, except the fact that joining uses trident tuple field from both sides to check and join two streams. Moreover, joining will work under batch level only. The sample code is as follows −
   
3. 
   
4. TridentTopology topology = new TridentTopology();
   
5. topology.merge(stream1, stream2, stream3);
   
6. topology.join(stream1, new Fields("key"), stream2, new Fields("x"),
   
7.    new Fields("key", "a", "b", "c"));

复制代码

1. Formatting the call information
   
2. The purpose of the FormatCall class is to format the call information comprising “Caller number” and “Receiver number”. The complete program code is as follows −
   
3. 
   
4. Coding: FormatCall.java
   
5. import backtype.storm.tuple.Values;
   
6. 
   
7. import storm.trident.operation.BaseFunction;
   
8. import storm.trident.operation.TridentCollector;
   
9. import storm.trident.tuple.TridentTuple;
   
10. 
    
11. public class FormatCall extends BaseFunction {
    
12.    @Override
    
13.    public void execute(TridentTuple tuple, TridentCollector collector) {
    
14.       String fromMobileNumber = tuple.getString(0);
    
15.       String toMobileNumber = tuple.getString(1);
    
16.       collector.emit(new Values(fromMobileNumber + " - " + toMobileNumber));
    
17.    }
    
18. }

复制代码

1. Hashtag Reader Bolt
   
2. The tweet emitted by spout will be forwarded to HashtagReaderBolt, which will process the tweet and emit all the available hashtags. HashtagReaderBolt uses getHashTagEntities method provided by twitter4j. getHashTagEntities reads the tweet and returns the list of hashtag. The complete program code is as follows −
   
3. 
   
4. Coding: HashtagReaderBolt.java
   
5. import java.util.HashMap;
   
6. import java.util.Map;
   
7. 
   
8. import twitter4j.*;
   
9. import twitter4j.conf.*;
   
10. 
    
11. import backtype.storm.tuple.Fields;
    
12. import backtype.storm.tuple.Values;
    
13. 
    
14. import backtype.storm.task.OutputCollector;
    
15. import backtype.storm.task.TopologyContext;
    
16. import backtype.storm.topology.IRichBolt;
    
17. import backtype.storm.topology.OutputFieldsDeclarer;
    
18. import backtype.storm.tuple.Tuple;
    
19. 
    
20. public class HashtagReaderBolt implements IRichBolt {
    
21.    private OutputCollector collector;
    
22. 
    
23.    @Override
    
24.    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
    
25.       this.collector = collector;
    
26.    }
    
27. 
    
28.    @Override
    
29.    public void execute(Tuple tuple) {
    
30.       Status tweet = (Status) tuple.getValueByField("tweet");
    
31.       for(HashtagEntity hashtage : tweet.getHashtagEntities()) {
    
32.          System.out.println("Hashtag: " + hashtage.getText());
    
33.          this.collector.emit(new Values(hashtage.getText()));
    
34.       }
    
35.    }
    
36. 
    
37.    @Override
    
38.    public void cleanup() {}
    
39. 
    
40.    @Override
    
41.    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
42.       declarer.declare(new Fields("hashtag"));
    
43.    }
    
44.        
    
45.    @Override
    
46.    public Map<String, Object> getComponentConfiguration() {
    
47.       return null;
    
48.    }
    
49.        
    
50. }

复制代码

1. Coding: HashtagCounterBolt.java
   
2. import java.util.HashMap;
   
3. import java.util.Map;
   
4. 
   
5. import backtype.storm.tuple.Fields;
   
6. import backtype.storm.tuple.Values;
   
7. 
   
8. import backtype.storm.task.OutputCollector;
   
9. import backtype.storm.task.TopologyContext;
   
10. 
    
11. import backtype.storm.topology.IRichBolt;
    
12. import backtype.storm.topology.OutputFieldsDeclarer;
    
13. import backtype.storm.tuple.Tuple;
    
14. 
    
15. public class HashtagCounterBolt implements IRichBolt {
    
16.    Map<String, Integer> counterMap;
    
17.    private OutputCollector collector;
    
18. 
    
19.    @Override
    
20.    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
    
21.       this.counterMap = new HashMap<String, Integer>();
    
22.       this.collector = collector;
    
23.    }
    
24. 
    
25.    @Override
    
26.    public void execute(Tuple tuple) {
    
27.       String key = tuple.getString(0);
    
28. 
    
29.       if(!counterMap.containsKey(key)){
    
30.          counterMap.put(key, 1);
    
31.       }else{
    
32.          Integer c = counterMap.get(key) + 1;
    
33.          counterMap.put(key, c);
    
34.       }
    
35.                
    
36.       collector.ack(tuple);
    
37.    }
    
38. 
    
39.    @Override
    
40.    public void cleanup() {
    
41.       for(Map.Entry<String, Integer> entry:counterMap.entrySet()){
    
42.          System.out.println("Result: " + entry.getKey()+" : " + entry.getValue());
    
43.       }
    
44.    }
    
45. 
    
46.    @Override
    
47.    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
48.       declarer.declare(new Fields("hashtag"));
    
49.    }
    
50.        
    
51.    @Override
    
52.    public Map<String, Object> getComponentConfiguration() {
    
53.       return null;
    
54.    }
    
55.        
    
56. }

复制代码

1. Submitting a Topology
   
2. Submitting a topology is the main application. Twitter topology consists of TwitterSampleSpout, HashtagReaderBolt, and HashtagCounterBolt. The following program code shows how to submit a topology.
   
3. 
   
4. Coding: TwitterHashtagStorm.java
   
5. import java.util.*;
   
6. 
   
7. import backtype.storm.tuple.Fields;
   
8. import backtype.storm.tuple.Values;
   
9. import backtype.storm.Config;
   
10. import backtype.storm.LocalCluster;
    
11. import backtype.storm.topology.TopologyBuilder;
    
12. 
    
13. public class TwitterHashtagStorm {
    
14.    public static void main(String[] args) throws Exception{
    
15.       String consumerKey = args[0];
    
16.       String consumerSecret = args[1];
    
17.                
    
18.       String accessToken = args[2];
    
19.       String accessTokenSecret = args[3];
    
20.                
    
21.       String[] arguments = args.clone();
    
22.       String[] keyWords = Arrays.copyOfRange(arguments, 4, arguments.length);
    
23.                
    
24.       Config config = new Config();
    
25.       config.setDebug(true);
    
26.                
    
27.       TopologyBuilder builder = new TopologyBuilder();
    
28.       builder.setSpout("twitter-spout", new TwitterSampleSpout(consumerKey,
    
29.          consumerSecret, accessToken, accessTokenSecret, keyWords));
    
30. 
    
31.       builder.setBolt("twitter-hashtag-reader-bolt", new HashtagReaderBolt())
    
32.          .shuffleGrouping("twitter-spout");
    
33. 
    
34.       builder.setBolt("twitter-hashtag-counter-bolt", new HashtagCounterBolt())
    
35.          .fieldsGrouping("twitter-hashtag-reader-bolt", new Fields("hashtag"));
    
36.                        
    
37.       LocalCluster cluster = new LocalCluster();
    
38.       cluster.submitTopology("TwitterHashtagStorm", config,
    
39.          builder.createTopology());
    
40.       Thread.sleep(10000);
    
41.       cluster.shutdown();
    
42.    }
    
43. }

复制代码

1. Submitting a Topology
   
2. This is the main application where YahooFinanceSpout.java and PriceCutOffBolt.java are connected together and produce a topology. The following program code shows how you can submit a topology.
   
3. 
   
4. Coding: YahooFinanceStorm.java
   
5. import backtype.storm.tuple.Fields;
   
6. import backtype.storm.tuple.Values;
   
7. 
   
8. import backtype.storm.Config;
   
9. import backtype.storm.LocalCluster;
   
10. import backtype.storm.topology.TopologyBuilder;
    
11. 
    
12. public class YahooFinanceStorm {
    
13.    public static void main(String[] args) throws Exception{
    
14.       Config config = new Config();
    
15.       config.setDebug(true);
    
16.                
    
17.       TopologyBuilder builder = new TopologyBuilder();
    
18.       builder.setSpout("yahoo-finance-spout", new YahooFinanceSpout());
    
19. 
    
20.       builder.setBolt("price-cutoff-bolt", new PriceCutOffBolt())
    
21.          .fieldsGrouping("yahoo-finance-spout", new Fields("company"));
    
22.                        
    
23.       LocalCluster cluster = new LocalCluster();
    
24.       cluster.submitTopology("YahooFinanceStorm", config, builder.createTopology());
    
25.       Thread.sleep(10000);
    
26.       cluster.shutdown();
    
27.    }
    
28. }

复制代码

多谢分享！