聊聊storm的direct grouping

本文主要研究一下storm的direct grouping

direct grouping

direct grouping是一种特殊的grouping,它是由上游的producer直接指定下游哪个task去接收它发射出来的tuple。direct grouping的使用有如下几个步骤:

1、上游在prepare方法保存下游bolt的taskId列表

public class SentenceDirectBolt extends BaseRichBolt {
    private static final Logger LOGGER = LoggerFactory.getLogger(SentenceDirectBolt.class);
    private OutputCollector collector;
    private List<Integer> taskIds;
    private int numCounterTasks;
    public void prepare(Map config, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        //NOTE 1 这里要取到下游的bolt的taskId,用于emitDirect时指定taskId
        this.taskIds = context.getComponentTasks("count-bolt");
        this.numCounterTasks = taskIds.size();
    }
    //......
}

这里保存了下游的bolt的taskId列表,用于emitDirect时选择taskId

2、上游在declareOutputFields使用declareStream声明streamId

public class SentenceDirectBolt extends BaseRichBolt {
    //......
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
        //NOTE 2 这里要通过declareStream声明direct stream,并指定streamId
        declarer.declareStream("directStreamDemo1",true,new Fields("word"));
        declarer.declareStream("directStreamDemo2",true,new Fields("word"));
    }
}

这里声明了两个streamId,一个是directStreamDemo1,一个是directStreamDemo2

3、上游采用emitDirect指定下游taskId及streamId

public class SentenceDirectBolt extends BaseRichBolt {
    //......
    public void execute(Tuple tuple) {
        String sentence = tuple.getStringByField("sentence");
        String[] words = sentence.split(" ");
        for(String word : words){
            int targetTaskId = getWordCountTaskId(word);
            LOGGER.info("word:{} choose taskId:{}",word,targetTaskId);
            // NOTE 3 这里指定发送给下游bolt的哪个taskId,同时指定streamId
            if(targetTaskId % 2 == 0){
                this.collector.emitDirect(targetTaskId,"directStreamDemo1",new Values(word));
            }else{
                this.collector.emitDirect(targetTaskId,"directStreamDemo2",new Values(word));
            }
        }
        this.collector.ack(tuple);
    }
}

这里使用emitDirect(int taskId, String streamId, List<Object> tuple)方法指定了下游的taskId以及要发送到的streamId

4、下游使用directGrouping连接上游bolt及streamId

    @Test
    public void testDirectGrouping() throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("sentence-spout", new SentenceSpout());
        // SentenceSpout --> SplitSentenceBolt
        builder.setBolt("split-bolt", new SentenceDirectBolt()).shuffleGrouping("sentence-spout");
        // SplitSentenceBolt --> WordCountBolt
        //NOTE 4这里要指定上游的bolt以及要处理的streamId
        builder.setBolt("count-bolt", new WordCountBolt(),5).directGrouping("split-bolt","directStreamDemo1");
        // WordCountBolt --> ReportBolt
        builder.setBolt("report-bolt", new ReportBolt()).globalGrouping("count-bolt");
        submitRemote(builder);
    }

这里count-bolt作为split-bolt的下游,使用了directGrouping,同时指定了要接收的streamId为directStreamDemo1

小结

  • direct grouping是一种特殊的grouping,它是由上游的producer直接指定下游哪个task去接收它发射出来的tuple。
  • 下游使用directGrouping连接上游同时指定要消费的streamId,上游在prepare的时候保存下游的taskId列表,然后在declareOutputFields的时候使用declareStream来声明streamId,最后在execute方法里头使用emitDirect(int taskId, String streamId, List tuple)方法指定了下游的taskId以及要发送到的streamId

doc

  • Concepts
  • Common Topology Patterns
  • 关于Storm Stream grouping

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏ASP.NET MVC5 后台权限管理系统

ASP.NET MVC5+EF6+EasyUI 后台管理系统(6)-Unity 依赖注入

前言 为了符合后面更新后的重构系统,文章于2016-11-1日重写 本节重构一下代码,采用IOC控制反转,也就是依赖注入 您可以访问http://unit...

29970
来自专栏菩提树下的杨过

Pechkin:html -> pdf 利器

Pechkin 是GitHub上的一个开源项目,可方便将html转化成pdf文档,使用也很方便,下面是winform项目中的示例代码: using System...

27250
来自专栏知识分享

51采集PCF8591数据通过ESP8266上传C#上位机android 之TCP客户端编程ESP8266使用详解NodeMCU初探ESP8266刷AT固件与nodemcu固件ESP8266使用详解-

这两天测试程序还发现一个bug就是如果客户端断开了,应该检测一下哪个断开了,数据就不应该发向那个连接,,,否则就会报错,然后模块会复位重启 所以加上这段代码 c...

59650
来自专栏ASP.NET MVC5 后台权限管理系统

ASP.NET MVC5+EF6+EasyUI 后台管理系统(63)-Excel导入和导出

昨天文章太过仓促没有补充导出的示例源码,在者当时弄到到很晚没时间做出导出功能,对阅读理解造成影响,现补充一份示例源码,顺便补充导出的功能说明,望理解 示例代...

53170
来自专栏码匠的流水账

聊聊storm的direct grouping

direct grouping是一种特殊的grouping,它是由上游的producer直接指定下游哪个task去接收它发射出来的tuple。direct gr...

11510
来自专栏柠檬先生

Angularjs基础(四)

AngularJS过滤器     过滤器可以使用一个管道符(|)添加到表达式和指令中。       AngularJS过滤器可用于转换数据:    ...

21290
来自专栏JackieZheng

Java豆瓣电影爬虫——模拟登录的前世今生与验证码的爱恨情仇

前言 并不是所有的网站都能够敞开心扉让你看个透彻,它们总要给你出些难题让你觉得有些东西是来之不易的,往往,这也更加激发你的激情和斗志! 从《为了媳妇的一张号,...

48960
来自专栏程序员的SOD蜜

一行代码调用实现带字段选取+条件判断+排序+分页功能的增强ORM框架

问题:3行代码 PDF.NET 是一个开源的数据开发框架,它的特点是简单、轻量、快速,易上手,而且是一个注释完善的国产开发框架,受到不少朋友的欢迎,也在我们公...

31690
来自专栏程序员与猫

EntityFramework Core 自动绑定模型映射

笔者最近在和同事共同开发项目时,需要从他们提供的包含数据库实体类型的类库中读取实体信息绑定到自己的项目中(但是都在同一个解决方案里),所以很直接的一种方式就是把...

24660
来自专栏张高兴的博客

张高兴的 Windows 10 IoT 开发笔记:HC-SR04 超声波测距模块

31070

扫码关注云+社区

领取腾讯云代金券