Storm(七)通过代码提交Topology

Storm(五)第一个集群Topology中用如下的命令行向集群提交了Topology:
storm jar cluster-topology-1.0-SNAPSHOT.jar com.quiterr.ExclamationTopology
但是在项目中需要动态自动化提交Topology,也就是通过代码来提交Topology。

这里仍然以ExclamationTopology为例来说明,不过要对ExclamationTopology做一些改造,要把ExclamationTopology中的Bolt单独提出来做一个项目(或模块),创建Topology相关的代码做成另一个项目。

一、Topology项目(后续以项目名dynamic-submit-classes来叙述)

这个项目用来放Topology自身相关的东西,包括Spout、Bolt等,实际项目中可以很复杂,但是在这个例子中很简单,因为Spout是Storm核心库提供的,而Bolt只有一个。
(一)pom文件

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

极简单,没什么好说的。
(二)ExclamationBolt
这个项目只有一个类ExclamationBolt,从ExclamationTopology里边copy出来的。

public class ExclamationBolt extends BaseRichBolt {
    OutputCollector _collector;

    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        _collector = collector;
    }

    public void execute(Tuple tuple) {
        _collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
        _collector.ack(tuple);
    }

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

}

(三)安装到本地
使用mvn install命令安装到本地,因为下一个项目要用到本项目。

(四)部署到Storm服务器
把生成的jar包放到Storm服务器上,这里放的路径是/home/app/test

本项目源代码:https://github.com/quiterr/storm-test/tree/master/dynamic-submit-classes

二、创建Topology项目(后续以项目名dynamic-submit来叙述)

用Spring Boot做一个RESTful API,专门用来接收创建Topology的请求,创建完毕后向集群提交Topology。
(一)pom文件

<dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>1.5.3.RELEASE</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--配置处理器,会在编辑配置文件的时候智能提示-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
        </dependency>
        <!--自己实现的topology所在的库-->
        <dependency>
            <groupId>com.quiterr</groupId>
            <artifactId>dynamic-submit-classes</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>1.5.3.RELEASE</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

除了Spring Boot相关的依赖,就是上一个项目的依赖。
(二)SubmitController
SubmitController接收一个topologyName参数,然后创建topology,最后提交到集群。代码中有一行指明了topology所在的位置:System.setProperty("storm.jar", "/home/app/test/dynamic-submit-classes-1.0-SNAPSHOT.jar");

@RestController
public class SubmitController {
    @RequestMapping(value = "/topology/create", method = RequestMethod.POST)
    private boolean create(@RequestBody String topologyName){
        Config conf = new Config();
        conf.setDebug(true);
        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");

        System.setProperty("storm.jar", "/home/app/test/dynamic-submit-classes-1.0-SNAPSHOT.jar");
        try {
            StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
        } catch (AlreadyAliveException e) {
            e.printStackTrace();
        } catch (InvalidTopologyException e) {
            e.printStackTrace();
        } catch (AuthorizationException e) {
            e.printStackTrace();
        }
        return true;
    }
}

本项目源代码:https://github.com/quiterr/storm-test/tree/master/dynamic-submit

三、测试和说明

把dynamic-submit-1.0-SNAPSHOT.jar和dynamic-submit-classes-1.0-SNAPSHOT.jar都放到Storm服务器上后,用java -jar dynamic-submit-1.0-SNAPSHOT.jar启动第一个项目,然后用一个RESTful的测试客户端发请求即可。

有个问题:topology被提交到哪里了?
Storm基础(三)配置这篇文章里提到过,如果没有定义storm.yml文件,就会使用Storm核心库中自带的default.xml文件,默认情况下topology就会提交到本地,这就是一定要把程序放到Storm所在的服务器上的原因。

四、用代码远程提交

能不能不把topology传到服务器上,就在本地提交呢?
理论上是可以的。
1.把dynamic-submit-classes-1.0-SNAPSHOT.jar放到dynamic-submit工程目录下。
2.StormSubmitter类中设置环境变量:System.setProperty("storm.jar", "dynamic-submit-classes-1.0-SNAPSHOT.jar");
3.dynamic-submit加一个storm.yml配置文件,内容像这样:

########### These MUST be filled in for a storm configuration
 storm.zookeeper.servers:
     - "172.19.3.147"
     - "172.19.3.148"
     - "172.19.3.149"
 
 nimbus.seeds: ["172.19.3.147"]
 storm.local.dir: "/data/storm_local"
 supervisor.slots.ports:
     - 6700
     - 6701
     - 6702
     - 6703
     - 6704
     - 6705
     - 6706
     - 6707
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,684评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 87,143评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,214评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,788评论 1 277
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,796评论 5 368
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,665评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,027评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,679评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 41,346评论 1 299
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,664评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,766评论 1 331
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,412评论 4 321
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,015评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,974评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,203评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,073评论 2 350
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,501评论 2 343

推荐阅读更多精彩内容