JStorm环境搭建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

package storm.cookbook;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;
/**
* Author: ashrith
* Date: 8/26/13
* Time: 12:03 PM
* Desc: setup the topology and submit it to either a local of remote Storm cluster depending on the arguments
*       passed to the main method.
*/
public class HelloWorldTopology {
/*
* main class in which to define the topology and a LocalCluster object (enables you to test and debug the
* topology locally). In conjunction with the Config object, LocalCluster allows you to try out different
* cluster configurations.
*
* Create a topology using 'TopologyBuilder' (which will tell storm how the nodes area arranged and how they
* exchange data)
* The spout and the bolts are connected using 'ShuffleGroupings'
*
* Create a 'Config' object containing the topology configuration, which is merged with the cluster configuration
* at runtime and sent to all nodes with the prepare method
*
* Create and run the topology using 'createTopology' and 'submitTopology'
*/
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("randomHelloWorld", new HelloWorldSpout(), 10);
builder.setBolt("HelloWorldBolt", new HelloWorldBolt(), 1).shuffleGrouping("randomHelloWorld");
Config conf = new Config();
conf.put(Config.NIMBUS_HOST, "localhost");
conf.put(Config.NIMBUS_THRIFT_PORT, 6627);
conf.setDebug(true);
if(args!=null && args.length > 0) {
conf.setNumWorkers(3);
StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
} else {
}
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
4445
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

package storm.cookbook;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import java.util.Map;
import java.util.Random;
/**
* Author: ashrith
* Date: 8/21/13
* Time: 8:33 PM
* Desc: spout essentially emits a stream containing 1 of 2 sentences 'Other Random Word' or 'Hello World' based on
*       random probability. It works by generating a random number upon construction and then generating subsequent
*       random numbers to test against the original member variable's value. When it matches "Hello World" is emitted,
*       during the remaining executions the other sentence is emitted.
*/
public class HelloWorldSpout extends BaseRichSpout{
private SpoutOutputCollector collector;
private int referenceRandom;
private static final int MAX_RANDOM = 10;
public HelloWorldSpout() {
final Random rand = new Random();
referenceRandom = rand.nextInt(MAX_RANDOM);
}
/*
* declareOutputFields() => you need to tell the Storm cluster which fields this Spout emits within the
*  declareOutputFields method.
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("sentence"));
}
/*
* open() => The first method called in any spout is 'open'
*           TopologyContext => contains all our topology data
*           SpoutOutputCollector => enables us to emit the data that will be processed by the bolts
*           conf => created in the topology definition
*/
@Override
public void open(Map conf, TopologyContext topologyContext, SpoutOutputCollector collector) {
this.collector = collector;
}
/*
* nextTuple() => Storm cluster will repeatedly call the nextTuple method which will do all the work of the spout.
*  nextTuple() must release the control of the thread when there is no work to do so that the other methods have
*  a chance to be called.
*/
@Override
public void nextTuple() {
final Random rand = new Random();
int instanceRandom = rand.nextInt(MAX_RANDOM);
if(instanceRandom == referenceRandom){
collector.emit(new Values("Hello World"));
} else {
collector.emit(new Values("Other Random Word"));
}
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

package storm.cookbook;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
import java.util.Map;
/**
* Author: ashrith
* Date: 8/26/13
* Time: 11:48 AM
* Desc: This bolt will consume the produced Tuples from HelloWorldSpout and implement the required counting logic
*/
public class HelloWorldBolt extends BaseRichBolt {
private int myCount = 0;
/*
* prepare() => on create
*/
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
}
/*
* execute() => most important method in the bolt is execute(Tuple input), which is called once per tuple received
*  the bolt may emit several tuples for each tuple received
*/
@Override
public void execute(Tuple tuple) {
String test = tuple.getStringByField("sentence");
if(test == "Hello World"){
myCount++;
System.out.println("Found a Hello World! My Count is now: " + Integer.toString(myCount));
}
}
/*
* declareOutputFields => This bolt emits nothing hence no body for declareOutputFields()
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
}
}