美文网首页Storm
Storm(四) 第一个Topology

Storm(四) 第一个Topology

作者: quiterr | 来源:发表于2017-04-12 09:41 被阅读90次

    之前已经翻译了好几篇官方的文档,项目第一阶段这两天也已经完成,按理说自己对Storm已经有一定的了解,但是由于是接手同事的工作,感觉Storm的基础没有打牢,所以自己来动手做做。

    原创文章,转载请注明出处

    官方推荐通过storm-starter来学习Storm,我也翻译了一篇文章:Storm(三) storm-starter,不过由于storm-starter包含多个Topology,做了一些抽象,本人觉得对初学者还是有点难度,所以决定自己新建项目。Storm(三) storm-starter这篇文章里提到了,对于初学者建议先看ExclamationTopology,我就从这个开始。如果对本文提到的Storm的概念不熟悉,建议先看Storm(二)官方Tutorial

    pom文件引入依赖

    既然是最简单的例子,那么依赖也是极简单的,只有一个:

    <dependencies>
            <dependency>
                <groupId>org.apache.storm</groupId>
                <artifactId>storm-core</artifactId>
                <version>1.0.1</version>
                <scope>provided</scope>
            </dependency>
    </dependencies>
    

    使用1.0.1版本是为了和服务器上部署的Storm集群保持一致,scope标签值设为provided,因为服务器上有运行环境,不需要生成的时候再包含。

    ExclamationBolt 的实现

      public static class ExclamationBolt extends BaseRichBolt {
        OutputCollector _collector;
    
        @Override
        public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
          _collector = collector;
        }
    
        @Override
        public void execute(Tuple tuple) {
          _collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
          _collector.ack(tuple);
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("word"));
        }
    
      }
    

    ExclamationBolt的代码和storm-starter里面是一致的,我没有做任何改动,这个bolt就是取出Tuple中的单词,在单词后面加"!!!",最后emit。

    main函数的实现

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        TopologyBuilder builder = new TopologyBuilder();
    
        builder.setSpout("word", new TestWordSpout(), 10);
        builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word");
        builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");
    
        conf.setDebug(true);
    
        String topologyName = "ExclamationTopology";
    
        try {
          LocalCluster cluster = new LocalCluster();
          cluster.submitTopology(topologyName, conf,builder.createTopology());
          Thread.sleep(60 * 1000);
          cluster.shutdown();
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
    

    我实现的main函数和storm-starter里面ExclamationTopology类的main函数大不一样,这里没有配置文件、没有参数解析,总之更简单些。数据来源是TestWordSpout,这是storm-core库里面的类,看一下源代码就知道它只是从5个人名中每次随机的发出1个:

    public void nextTuple() {
            Utils.sleep(100);
            final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"};
            final Random rand = new Random();
            final String word = words[rand.nextInt(words.length)];
            _collector.emit(new Values(word));
        }
    

    数据处理就是上文提到的ExclamationBolt,每次加"!!!",然后emit。
    这个Topology就是一个三个节点串在一起:word->exclaim1->exclaim2,word的并发度是10,exclaim1的并发度是3,exclaim2的并发度是2,并发度对应线程的概念。分组方式是随机分组(shuffleGrouping)。
    为了简单,我们让它以local模式运行,60秒后自动停止。

    打包&运行

    我用的IDEA,如果是命令,可以这样打包:
    mvn package
    像这样提交到Storm运行:

    storm jar first-topology-1.0-SNAPSHOT.jar com.quiterr.ExclamationTopology
    

    first-topology是项目的模块名,com.quiterr是Java 包名,请做相应更改。

    在IDEA中运行

    Storm的local模式已经比集群模式简单多了,通常在项目开发、测试阶段使用,但是我们还想更方便一些,比如在IDEA中运行和调试Topology。只需要把依赖的scope那一行去掉即可(提供storm的本地运行环境):

    <dependencies>
            <dependency>
                <groupId>org.apache.storm</groupId>
                <artifactId>storm-core</artifactId>
                <version>1.0.1</version>
                <!--<scope>provided</scope>-->
            </dependency>
    </dependencies>
    

    很简单吧,这样就可以在IDEA中使用断点之类的强大功能了。

    本文源代码:https://github.com/quiterr/storm-test/tree/master/first-topology

    相关文章

      网友评论

        本文标题:Storm(四) 第一个Topology

        本文链接:https://www.haomeiwen.com/subject/mtruattx.html