Storm 运行例子

1.建立Java工程

使用idea，添加lib库，拷贝storm中lib到工程中

2.拷贝wordcount代码

下载src包，解压找到

apache-storm-0.9.4-srcapache-storm-0.9.4examplesstorm-startersrcjvmstormstarter目录下

拷贝WordCountTopology.java内容；

修改python处理方式；

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.task.ShellBolt;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.bigdata.storm.spout.*;

import java.util.HashMap;
import java.util.Map;
/**
 * Created by Edward on 2016/8/17.
 */
public class MyTest {

    public static class SplitSentence extends BaseBasicBolt{

        @Override
        public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
            String word = tuple.getString(0);
            String str[] = word.split(" ");

            System.out.println("Split Sentence:" + tuple.getSourceStreamId());
            for(int i=0; i<str.length; i++)
            {
                basicOutputCollector.emit(new Values(str[i]));
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("word"));
        }
    }

    public static class WordCount extends BaseBasicBolt {
        Map<String, Integer> counts = new HashMap<String, Integer>();

        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            String word = tuple.getString(0);
            Integer count = counts.get(word);
            if (count == null)
                count = 0;
            count++;
            counts.put(word, count);
            System.out.println("Word Count:" + tuple.getSourceStreamId());
            collector.emit(new Values(word, count));

        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word", "count"));
        }
    }

    public static void main(String[] args) throws Exception {

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("spout", new RandomSentenceSpout(), 5);

        builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout");
        builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));

        Config conf = new Config();
        conf.setDebug(true);


        if (args != null && args.length > 0) {
            conf.setNumWorkers(3);

            StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
        }
        else {
            conf.setMaxTaskParallelism(3);

            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("word-count", conf, builder.createTopology());

            Thread.sleep(50000);

            cluster.shutdown();
        }
    }
}

3.拷贝随机生成spout代码

找到 apache-storm-0.9.4-srcapache-storm-0.9.4examplesstorm-startersrcjvmstormstarterspout

拷贝RandomSentenceSpout.java到工程中

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 backtype.storm.utils.Utils;

import java.util.Map;
import java.util.Random;

public class RandomSentenceSpout extends BaseRichSpout {
    SpoutOutputCollector _collector;
    Random _rand;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        _collector = collector;
        _rand = new Random();
    }

    @Override
    public void nextTuple() {
        Utils.sleep(10000);
        String[] sentences = new String[]{ "the cow jumped over the moon", "an apple a day keeps the doctor away",
                "four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" };
        String sentence = sentences[_rand.nextInt(sentences.length)];
        Object id = new Object();
        System.out.println(id);
        //id message ID 用来保证可靠性的，如果失败fail 会返回 message id 信息
        _collector.emit(new Values(sentence), id);
    }

    @Override
    public void ack(Object id) {
        System.out.println("storm spout ack id = "+id);
    }

    @Override
    public void fail(Object id) {
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }

}

4.本地运行

在idea中直接点击运行，观察运行过程

5.集群运行

将程序打包成jar,然后放到集群中运行。

把jar包放到storm/test目录下,执行：

bin/storm jar test/storm-sample.jar com.bigdata.storm.MyTest WordCount