Storm

Storm(四) 第一个Topology

2017-04-12  本文已影响90人  quiterr

之前已经翻译了好几篇官方的文档,项目第一阶段这两天也已经完成,按理说自己对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

上一篇下一篇

猜你喜欢

热点阅读