Flux模式开发提交JStorm任务

原文地址查看本文原址

传统法式采用提交jar包的方式运行topology，一旦我们需要改变拓扑里头的相应配置，我们就必须重新编译和打包，而Flux可以帮助我们创建和部署jstorm拓扑的编程框架及组件。它可以将你代码中有关topology结构以及提交部分用一句话加上配置文件完成。

传统方式

在jar内完成topology的构建以及数据流配置，代码可能如下：

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("send",new genRandomSentenceSpout());
builder.setBolt("split",new splitSentenceBolt()).shuffleGrouping("send");
        builder.setBolt("count",new wordCountBolt()).fieldsGrouping("split",new Fields("word"));

Config conf=new Config();
conf.setNumWorkers(1);
conf.setNumAckers(1);

boolean runLocal = shouldRunLocal();
if(runLocal){
    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology(name, conf, builder.createTopology());    //本地提交
} else {
    StormSubmitter.submitTopology(name, conf, builder.createTopology());  //集群提交
    }
}

使用Flux，上面代码可用如下Flux命令代替：

jstorm jar mytopology.jar com.alibaba.jstorm.flux.Flux --local config.yaml //本地提交
jstorm jar mytopology.jar com.alibaba.jstorm.flux.Flux --remote config.yaml //远程提交

Flux方式开发

maven依赖与打包配置

由于需要maven依赖flux-core，而flux-core在网上没有链接可以下载，所以需要手动生产安装。通过集群版本下载对应JStorm源码，maven中编译安装JStorm-Flux，会在你本地maven仓库中安装jstorm-core.jar。

编译安装JStorm-Flux

然后在开发topology项目中添加maven依赖：

<dependencies>
        <dependency>
            <groupId>com.alibaba.jstorm</groupId>
            <artifactId>flux-core</artifactId>
            <version>2.2.1</version>
        </dependency>
</dependencies>

如下代码以maven-shade打包为例，在pom.xml中添加打包方式，其中mainClass设置为com.alibaba.jstorm.flux.Flux

<build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <configuration>
                    <createDependencyReducedPom>true</createDependencyReducedPom>
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.alibaba.jstorm.flux.Flux</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

配置文件

开发完spout、bolt后不需要在main函数中显示配置topology的结构，采用配置文件的方式来构建topology结构。例如如下的代码跟配置文件在效果上是一样的。

//代码方式构建topology
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("send",new genRandomSentenceSpout());
builder.setBolt("split",new splitSentenceBolt()).shuffleGrouping("send");
        builder.setBolt("count",new wordCountBolt()).fieldsGrouping("split",new Fields("word"));

Config conf=new Config();
conf.setNumWorkers(1);
conf.setNumAckers(1);

StormSubmitter. submitTopology(topo_name , conf, builder.createTopology() );

# Flux配置文件方式
---
# 定义topology名
name: "flux"
# topology有关配置，worker、acker数量配置
config:
  topology.workers: 1
  topology.ackers: 1
# spouts配置
spouts:
  - id: "word-spout"
    className: "spout.genRandomSentenceSpout"
parallelism: 1
# Bolt配置
bolts:
  - id: "word-counter"
    className: "bolt.wordCountBolt"
    parallelism: 1

  - id: "split-bolt"
    className: "bolt.splitSentenceBolt"
    parallelism: 1
# 数据流配置
streams:
  - name: "word-spout --> split-bolt" # name isn't used (placeholder for logging, UI, etc.)
    from: "word-spout"
    to: "split-bolt"
    grouping:
      type: SHUFFLE

  - name: "split-bolt --> word-counter"
    from: "split-bolt"
    to: "word-counter"
    grouping:
      type: SHUFFLE
      args: ["word"]

发布提交

一旦你用flux完成了topology打包，你就可以利用配置文件来跑各种拓扑啦。比如你的jar名称为myTopology-0.1.0-SNAPSHOT.jar， 你可以利用以下命令跑本地模式

jstorm jar myTopology-0.1.0-SNAPSHOT.jar com.alibaba.jstorm.flux.Flux --local my_config.yaml

当然你也可以跑分布式模式

jstorm jar myTopology-0.1.0-SNAPSHOT.jar com.alibaba.jstorm.flux.Flux --remote my_config.yaml