Apache Storm Tutorial

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1. Apache Storm Tutorial
   
2. Apache Storm - Home
   
3. Apache Storm - Introduction
   
4. Apache Storm - Core Concepts
   
5. Apache Storm - Cluster Architecture
   
6. Apache Storm - Workflow
   
7. Storm - Distributed Msging System
   
8. Apache Storm - Installation
   
9. Apache Storm - Working Example
   
10. Apache Storm - Trident
    
11. Apache Storm in Twitter
    
12. Apache Storm in Yahoo! Finance
    
13. Apache Storm - Applications
    
14. Apache Storm Useful Resources
    
15. Apache Storm - Quick Guide
    
16. Apache Storm - Useful Resources
    
17. Apache Storm - Discussion
    
18. Selected Reading
    
19. Developer's Best Practices
    
20. Questions and Answers
    
21. Effective Resume Writing
    
22. HR Interview Questions
    
23. Computer Glossary
    
24. Who is Who

复制代码

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Tutorial apache Storm tutor Tori

1. Coding − FakeCallLogReaderSpout.java
   
2. import java.util.*;
   
3. //import storm tuple packages
   
4. import backtype.storm.tuple.Fields;
   
5. import backtype.storm.tuple.Values;
   
6. 
   
7. //import Spout interface packages
   
8. import backtype.storm.topology.IRichSpout;
   
9. import backtype.storm.topology.OutputFieldsDeclarer;
   
10. import backtype.storm.spout.SpoutOutputCollector;
    
11. import backtype.storm.task.TopologyContext;
    
12. 
    
13. //Create a class FakeLogReaderSpout which implement IRichSpout interface
    
14.    to access functionalities
    
15.        
    
16. public class FakeCallLogReaderSpout implements IRichSpout {
    
17.    //Create instance for SpoutOutputCollector which passes tuples to bolt.
    
18.    private SpoutOutputCollector collector;
    
19.    private boolean completed = false;
    
20.        
    
21.    //Create instance for TopologyContext which contains topology data.
    
22.    private TopologyContext context;
    
23.        
    
24.    //Create instance for Random class.
    
25.    private Random randomGenerator = new Random();
    
26.    private Integer idx = 0;
    
27. 
    
28.    @Override
    
29.    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    
30.       this.context = context;
    
31.       this.collector = collector;
    
32.    }
    
33. 
    
34.    @Override
    
35.    public void nextTuple() {
    
36.       if(this.idx <= 1000) {
    
37.          List<String> mobileNumbers = new ArrayList<String>();
    
38.          mobileNumbers.add("1234123401");
    
39.          mobileNumbers.add("1234123402");
    
40.          mobileNumbers.add("1234123403");
    
41.          mobileNumbers.add("1234123404");
    
42. 
    
43.          Integer localIdx = 0;
    
44.          while(localIdx++ < 100 && this.idx++ < 1000) {
    
45.             String fromMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
    
46.             String toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
    
47.                                
    
48.             while(fromMobileNumber == toMobileNumber) {
    
49.                toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
    
50.             }
    
51.                                
    
52.             Integer duration = randomGenerator.nextInt(60);
    
53.             this.collector.emit(new Values(fromMobileNumber, toMobileNumber, duration));
    
54.          }
    
55.       }
    
56.    }
    
57. 
    
58.    @Override
    
59.    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
60.       declarer.declare(new Fields("from", "to", "duration"));
    
61.    }
    
62. 
    
63.    //Override all the interface methods
    
64.    @Override
    
65.    public void close() {}
    
66. 
    
67.    public boolean isDistributed() {
    
68.       return false;
    
69.    }
    
70. 
    
71.    @Override
    
72.    public void activate() {}
    
73. 
    
74.    @Override
    
75.    public void deactivate() {}
    
76. 
    
77.    @Override
    
78.    public void ack(Object msgId) {}
    
79. 
    
80.    @Override
    
81.    public void fail(Object msgId) {}
    
82. 
    
83.    @Override
    
84.    public Map<String, Object> getComponentConfiguration() {
    
85.       return null;
    
86.    }
    
87. }

复制代码

1. Coding − CallLogCreatorBolt.java
   
2. //import util packages
   
3. import java.util.HashMap;
   
4. import java.util.Map;
   
5. 
   
6. import backtype.storm.tuple.Fields;
   
7. import backtype.storm.tuple.Values;
   
8. import backtype.storm.task.OutputCollector;
   
9. import backtype.storm.task.TopologyContext;
   
10. 
    
11. //import Storm IRichBolt package
    
12. import backtype.storm.topology.IRichBolt;
    
13. import backtype.storm.topology.OutputFieldsDeclarer;
    
14. import backtype.storm.tuple.Tuple;
    
15. 
    
16. //Create a class CallLogCreatorBolt which implement IRichBolt interface
    
17. public class CallLogCreatorBolt implements IRichBolt {
    
18.    //Create instance for OutputCollector which collects and emits tuples to produce output
    
19.    private OutputCollector collector;
    
20. 
    
21.    @Override
    
22.    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
    
23.       this.collector = collector;
    
24.    }
    
25. 
    
26.    @Override
    
27.    public void execute(Tuple tuple) {
    
28.       String from = tuple.getString(0);
    
29.       String to = tuple.getString(1);
    
30.       Integer duration = tuple.getInteger(2);
    
31.       collector.emit(new Values(from + " - " + to, duration));
    
32.    }
    
33. 
    
34.    @Override
    
35.    public void cleanup() {}
    
36. 
    
37.    @Override
    
38.    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
39.       declarer.declare(new Fields("call", "duration"));
    
40.    }
    
41.        
    
42.    @Override
    
43.    public Map<String, Object> getComponentConfiguration() {
    
44.       return null;
    
45.    }
    
46. }

复制代码

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

复制代码

1. Creating Topology
   
2. The Storm topology is basically a Thrift structure. TopologyBuilder class provides simple and easy methods to create complex topologies. The TopologyBuilder class has methods to set spout (setSpout) and to set bolt (setBolt). Finally, TopologyBuilder has createTopology to create topology. Use the following code snippet to create a topology −
   
3. 
   
4. TopologyBuilder builder = new TopologyBuilder();
   
5. 
   
6. builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());
   
7. 
   
8. builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
   
9.    .shuffleGrouping("call-log-reader-spout");
   
10. 
    
11. builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
    
12.    .fieldsGrouping("call-log-creator-bolt", new Fields("call"));

复制代码

1. Coding − LogAnalyserStorm.java
   
2. import backtype.storm.tuple.Fields;
   
3. import backtype.storm.tuple.Values;
   
4. 
   
5. //import storm configuration packages
   
6. import backtype.storm.Config;
   
7. import backtype.storm.LocalCluster;
   
8. import backtype.storm.topology.TopologyBuilder;
   
9. 
   
10. //Create main class LogAnalyserStorm submit topology.
    
11. public class LogAnalyserStorm {
    
12.    public static void main(String[] args) throws Exception{
    
13.       //Create Config instance for cluster configuration
    
14.       Config config = new Config();
    
15.       config.setDebug(true);
    
16.                
    
17.       //
    
18.       TopologyBuilder builder = new TopologyBuilder();
    
19.       builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());
    
20. 
    
21.       builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
    
22.          .shuffleGrouping("call-log-reader-spout");
    
23. 
    
24.       builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
    
25.          .fieldsGrouping("call-log-creator-bolt", new Fields("call"));
    
26.                        
    
27.       LocalCluster cluster = new LocalCluster();
    
28.       cluster.submitTopology("LogAnalyserStorm", config, builder.createTopology());
    
29.       Thread.sleep(10000);
    
30.                
    
31.       //Stop the topology
    
32.                
    
33.       cluster.shutdown();
    
34.    }
    
35. }

复制代码

1. Python Binding
   
2. Python is a general-purpose interpreted, interactive, object-oriented, and high-level programming language. Storm supports Python to implement its topology. Python supports emitting, anchoring, acking, and logging operations.
   
3. 
   
4. As you know, bolts can be defined in any language. Bolts written in another language are executed as sub-processes, and Storm communicates with those sub-processes with JSON messages over stdin/stdout. First take a sample bolt WordCount that supports python binding.
   
5. 
   
6. public static class WordCount implements IRichBolt {
   
7.    public WordSplit() {
   
8.       super("python", "splitword.py");
   
9.    }
   
10.        
    
11.    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
12.       declarer.declare(new Fields("word"));
    
13.    }
    
14. }
    
15. Here the class WordCount implements the IRichBolt interface and running with python implementation specified super method argument "splitword.py". Now create a python implementation named "splitword.py".
    
16. 
    
17. import storm
    
18.    class WordCountBolt(storm.BasicBolt):
    
19.       def process(self, tup):
    
20.          words = tup.values[0].split(" ")
    
21.          for word in words:
    
22.          storm.emit([word])
    
23. WordCountBolt().run()

复制代码

1. Trident Spout
   
2. Trident spout is similar to Storm spout, with additional options to use the features of Trident. Actually, we can still use the IRichSpout, which we have used in Storm topology, but it will be non-transactional in nature and we won’t be able to use the advantages provided by Trident.
   
3. 
   
4. The basic spout having all the functionality to use the features of Trident is "ITridentSpout". It supports both transactional and opaque transactional semantics. The other spouts are IBatchSpout, IPartitionedTridentSpout, and IOpaquePartitionedTridentSpout.
   
5. 
   
6. In addition to these generic spouts, Trident has many sample implementation of trident spout. One of them is FeederBatchSpout spout, which we can use to send named list of trident tuples easily without worrying about batch processing, parallelism, etc.
   
7. 
   
8. FeederBatchSpout creation and data feeding can be done as shown below −
   
9. 
   
10. TridentTopology topology = new TridentTopology();
    
11. FeederBatchSpout testSpout = new FeederBatchSpout(
    
12.    ImmutableList.of("fromMobileNumber", "toMobileNumber", “duration”));
    
13. topology.newStream("fixed-batch-spout", testSpout)
    
14. testSpout.feed(ImmutableList.of(new Values("1234123401", "1234123402", 20)));

复制代码

1. Filter
   
2. Filter is an object used to perform the task of input validation. A Trident filter gets a subset of trident tuple fields as input and returns either true or false depending on whether certain conditions are satisfied or not. If true is returned, then the tuple is kept in the output stream; otherwise, the tuple is removed from the stream. Filter will basically inherit from the BaseFilter class and implement the isKeep method. Here is a sample implementation of filter operation −
   
3. 
   
4. public class MyFilter extends BaseFilter {
   
5.    public boolean isKeep(TridentTuple tuple) {
   
6.       return tuple.getInteger(1) % 2 == 0;
   
7.    }
   
8. }
   
9. 
   
10. input
    
11. 
    
12. [1, 2]
    
13. [1, 3]
    
14. [1, 4]
    
15. 
    
16. output
    
17. 
    
18. [1, 2]
    
19. [1, 4]

复制代码

1. Function basically inherits from the BaseFunction class and implements the execute method. A sample implementation is given below −
   
2. 
   
3. public class MyFunction extends BaseFunction {
   
4.    public void execute(TridentTuple tuple, TridentCollector collector) {
   
5.       int a = tuple.getInteger(0);
   
6.       int b = tuple.getInteger(1);
   
7.       collector.emit(new Values(a + b));
   
8.    }
   
9. }
   
10. 
    
11. input
    
12. 
    
13. [1, 2]
    
14. [1, 3]
    
15. [1, 4]
    
16. 
    
17. output
    
18. 
    
19. [1, 2, 3]
    
20. [1, 3, 4]
    
21. [1, 4, 5]
    
22. Just like Filter operation, Function operation can be called in a topology using the each method. The sample code is as follows −
    
23. 
    
24. TridentTopology topology = new TridentTopology();
    
25. topology.newStream("spout", spout)
    
26.    .each(new Fields(“a, b"), new MyFunction(), new Fields(“d")));

复制代码