皮皮网

1.如何在eclipse调试storm程序

如何在eclipse调试storm程序

       ä¸€ã€ä»‹ç»

        storm提供了两种运行模式：本地模式和分布式模式。本地模式针对开发调试storm topologies非常有用。

        Storm has two modes of operation: local mode and distributed mode.

        In local mode,qt源码 vs Storm executes completely in process by simulating

       worker nodes with threads. Local mode is useful for testing and

       development of topologies

        因为多数程序开发者都是使用windows系统进行程序开发，如果在本机不安装storm环境的情况下，开发、调试storm程序。如果你正在为此问题而烦恼，请使用本文提供的方法。

        二、实施步骤

        如何基于eclipse+maven调试storm程序，步骤如下：

        1.搭建好开发环境（eclipse+maven,本人使用的是eclipse Kepler 与maven3.1.1）

        2.创建maven项目，并修改pom.xml，内容如pom.xml（机器联网，下载所需的依赖jar）

        Github上的pom.xml,引入的依赖太多，有些不需要，

        3. 编写storm程序，指定为本地模式运行。本文提供的程序是wordcount

        重要的是LocalCluster cluster = new LocalCluster();这一句

       Config conf = new Config();

       conf.setDebug(true);

       conf.setNumWorkers(2);

       LocalCluster cluster = new LocalCluster();

       cluster.submitTopology("test", conf, builder.createTopology());

       Utils.sleep();

       cluster.killTopology("test");

       cluster.shutdown();

       pom.xml文件

       <project xmlns="mons-collections</groupId>

        <artifactId>commons-collections</artifactId>

        <version>3.2.1</version>

        </dependency>

        </dependencies>

       </project>

       storm程序

       package storm.starter;

       import java.util.HashMap;

       import java.util.Map;

       import storm.starter.spout.RandomSentenceSpout;

       import backtype.storm.Config;

       import backtype.storm.LocalCluster;

       import backtype.storm.StormSubmitter;

       import backtype.storm.topology.BasicOutputCollector;

       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;

       /

        * This topology demonstrates Storm's stream groupings and multilang

        * capabilities.

        */

       public class WordCountTopology {

        public static class SplitSentence extends BaseBasicBolt {

        @Override

        public void execute(Tuple input, BasicOutputCollector collector) {

        try {

        String msg = input.getString(0);

        System.out.println(msg + "-------------------");

        if (msg != null) {

        String[] s = msg.split(" ");

        for (String string : s) {

        collector.emit(new Values(string));

        }

        }

        } catch (Exception e) {

        e.printStackTrace();

        }

        }

        @Override

        public void declareOutputFields(OutputFieldsDeclarer declarer) {

        declarer.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);

        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(), ).fieldsGrouping("split",

        new Fields("word"));

        Config conf = new Config();

        conf.setDebug(true);

        if (args != null && args.length > 0) {

        conf.setNumWorkers(3);

        StormSubmitter.submitTopology(args[0], conf,

        builder.createTopology());

        } else {

        conf.setMaxTaskParallelism(3);

        LocalCluster cluster = new LocalCluster();

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

        Thread.sleep();

        cluster.shutdown();

        }

        }

       }

       package storm.starter.spout;

       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();

        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)];

        _collector.emit(new Values(sentence));

        }

        @Override

        public void ack(Object id) {

        }

        @Override

        public void fail(Object id) {

        }

        @Override

        public void declareOutputFields(OutputFieldsDeclarer declarer) {

        declarer.declare(new Fields("word"));

        }

       }