Storm Trident之一spout和bolt

Trident Spout


Trident Spout特点

  • Trident中,定义Spout的接口为ITridentSpout。
  • Trident Spout必须以批量形式发送tuple。
  • Trident Spout不真正执行数据的发送,而是由ITridentSpout.Emitter负责发送数据。同时引入了协调器的概念,协调器负责管理数据发送的批次和元数据,当事务失败时,调度Emitter根据元数据重新发送数据。协调器接口为ITridentSpout.BatchCoordinator。

maven依赖

<dependency>
    <groupId>org.apache.storm</groupId>
    <artifactId>storm-core</artifactId>
    <version>1.0.2</version>
</dependency>

Trident Spout简单实现

    public class WordSpout implements ITridentSpout<String> {
        
        /**
         * 
         */
        private static final long serialVersionUID = -954626449213280061L;
    
        /**
         * 协调器
         * 负责保存重放batch元数据,当重放一个batch时,通过协调器中保存的元数据创建batch
         */
        @Override
        public BatchCoordinator<String> getCoordinator(String txStateId,Map conf, TopologyContext context) {
            return new WordCoordinator();
        }
    
        @Override
        public Emitter<String> getEmitter(String txStateId, Map conf, TopologyContext context) {
            return new WordEmitter();
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    
        /**
         * 定义发送的所有字段
         */
        @Override
        public Fields getOutputFields() {
            return new Fields("field1","field2");
        }
        
        private class WordCoordinator implements BatchCoordinator<String> {
    
            @Override
            public String initializeTransaction(long txid, String prevMetadata, String currMetadata) {
                return null;
            }
    
            @Override
            public void success(long txid) {
                logger.info("success: " + txid);
            }
    
            @Override
            public boolean isReady(long txid) {
                return Boolean.TRUE;
            }
    
            @Override
            public void close() {
                
            }
            
        }
        
        /**
         * 发射器
         * 发送数据流
         *
         */
        private class WordEmitter implements Emitter<String> {
    
            @Override
            public void success(TransactionAttempt tx) {
                logger.info("emitter success " + tx.getId());
            }
    
            @Override
            public void close() {
            }
            
            /**
             * 每次调用本方法所发送的数据集合被称为batch
             * batch是Trident中发送数据流的最小单元
             */
            @Override
            public void emitBatch(TransactionAttempt tx, String coordinatorMeta, TridentCollector collector) {
                for(int i=0;i<10;i++){
                    List list = Lists.newArrayList();
                    list.add("event1");
                    list.add("event2");
                    collector.emit(list);
                }
            }
            
        }
    
        private Logger logger = LoggerFactory.getLogger("Trident Spout");
    }
重点说明

Emitter定义的emitBatch方法。该方法实现了发送哪里数据。该方法每执行一次,发送的所有数据被称为batch。batch中的每条数据被称为tuple。

ITridentSpout.getOutputFields定义了每条tuple有哪些字段,本例中定义了2个字段,字段名为"field1"、"field2"。在Emitter.emitBatch中每条tuple均符合该定义。本例中每次调用emitBatch方法发送的数据内容及格式可以假象如下:

[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]
[event1,event2]

每次调用均发送了10条数据(10个tuple),这10个tuple构成1个batch,tuple均符合ITridentSpout.getOutputFields中定义的字段

Trident bolt


Trident bolt特点

  • Trident中没有bolt接口,而是分为了Filter和Function两类

Trident Function简单实现

public class WordFunction extends BaseFunction {

    /**
     * 
     */
    private static final long serialVersionUID = 735468688795780833L;

    /**
     * 接收数据流
     * 每次接收batch中一条数据
     */
    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
        logger.info(tuple.getValueByField("field1").toString());
    }

    private Logger logger = LoggerFactory.getLogger("Trident Function");
}
重点说明

BaseFunction已经实现了Function接口。

execute方法用于具体实现接收到tuple后如何处理。每次接收1个tuple。在本例中emitter每次发送1个batch,每个batch有10条数据,则每次发送数据,execute方法均会被调用10次。

在tuple中可以获取数据流中的数据,能够获取的字段受TridentTopology对象的控制。

在execute方法中处理完成后可继续使用TridentCollector对象继续发送数据到下一节点,Function发送数据时只能添加新的字段,不能修改或删除已有的字段

启动TridentTopology


public class Start {

    public static StormTopology buildTopology() {
        TridentTopology topology = new TridentTopology();
        WordSpout spout = new WordSpout();
        WordFunction function = new WordFunction();

        topology.newStream("filter", spout)
                /**
                 * 将spout发送的数据流中哪些字段传入bolt中
                 */
                .each(new Fields("field1"), function, new Fields());

        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("MyStorm", conf, buildTopology());

        Thread.sleep(1000 * 60);
        cluster.shutdown();
    }

}
重点说明

topology定义数据流时指定function可以读取哪些字段。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • 一. wordCount Topology开发: 1.spout数据收集器(SentenceSpout类): 有...
    奉先阅读 1,223评论 0 0
  • 这是一个JStorm使用教程,不包含环境搭建教程,直接在公司现有集群上跑任务,关于JStorm集群环境搭建,后续研...
    Coselding阅读 6,453评论 1 9
  • 声明 本文首发于个人技术博客,转载请注明出处,本文链接:http://qifuguang.me/2015/11/2...
    winwill2012阅读 2,220评论 1 15
  • 简介: Trident 是 Storm 的一种高度抽象的实时计算模型,它可以将高吞吐量(每秒百万级)数据输入、有状...
    hello_coke阅读 3,286评论 0 1
  • 现在的社会太匆忙,人情关系太复杂,有些事情只能靠自己解决,无论是工作还是生活。有自己的想法,就应该想各种办法,最终...
    平静的海洋阅读 1,563评论 0 0