Storm Topology及分组整理

Storm的通信机制,需要满足如下一些条件以满足Storm的语义。

1、建立数据传输的缓冲区。在通信连接没有建立之前把发送的数据缓存起来。数据发送方可以在连接建立之前发送消息,而不需要等连接建立起来,可是的接收方是独立运行的。

2、在消息传输层保证消息最多只能发送一次,Storm系统有ACK机制,是的没有被发送成功的消息会被重发,若消息层面也重发,会导致消息发送多次。

这种消息机制由两个接口来定义,backtype.storm.messaging.IContext和backtype.storm.messaging.IConnection.

IContext负责客户端和服务器端建立的连接,主要有四个方法。

1、prepare(Map stormConf):总从Storm定义的prepare方法,可以接收storm的配置。

2、term():终止,方法会在worker卸载这个传输插件的时候调用,自定义实现时可以在这里释放占用的资源。

3、bind(String topologyId,int port):建立服务器端的连接。

4、connect(String stormId,String host,int port):建立一个客户端的连接。

IConnect定义了在IContext上发送、接收数据的接口。

1、recv(int flag):接收消息。

2、send(int taskId,byte[] payload):发送消息。

3、close():该连接关闭的时候调用,释放相关资源。

Topology原理整理

  从运行时Topology的实际执行过程角度,作业是由多个组件的实例,也即任务,按照构造时简历的逻辑顺序呢和配置的并发度,形成的数据流图结构。

  流(stream)是Storm中对传递的数据进行的抽象,流是时间上无限的数据项Tuple序列。Spout是Stream的源,为Topology从特定数据源获取数据项,并向作业中发射(emit)形成Stream。(项目中使用了kafkaspout,接收后进行数据校验再使用emit发送给bolt),bolt可以同时接受任意多个上游送达的Stream作为输入,进行数据的处理过程,也可以在bolt做完处理后执行(emit)发射新的Stream继续给下游的Bolt进行处理。

  Stream中的Tuple可以被指定结构,由一个或多个域(field)组成。Tuple的定义不必是严格统一的,而是可以在每个spout,bolt中定义。默认情况下Tuple可以包含基本类型,如integers、longs、shorts、bytes、strings、doubles、floats、booleans和byte arrays.

流组模式

1、Shuffle Grouping 随机分组

public void createTopology(TopologyBuilder builder){
    kafkaSpout kafkaspout = getKafkaSpout(topicName);
    //Topology中增加一个Spout
    builder.setSpout(...)
    //在Topology中增加一个Bolt,可设置并行度,以随机分组的方式发送,shuffleGrouping后的参数为源组建的Id
    builder.setBolet(boltName,new BlackListBolt(),3).shuffleGrouping(spoutName);}

在这种流组模式下,源组件将其发送的数据项,以随机的方式向其所有目标组件发送,可以保证每个目标组件收到数量近似的Tuple。

2、All Grouping 副本分组

//allGrouping(java.lang.String componentId)
//allGrouping(java.lang.String componentId,java.lang.String streamId)
//参数streamId是声明的流的标识
builder.setBolet(boltName,new BlackListBolt(),3).allGrouping(spoutName,"signals“);

在这种模式下,源组件将其发送的数据项,以副本的形式向其所有目标组件发送,可以保证每个目标组件均收到同一个Tuple,就好比zookeeper的配置文件同步一样,每个bolt都会收到同一份。

3、Global Grouping 全局分组

这种模式下,源组件将其发送的数据项,全部发送给目标组件的某一个实例,而且该实例是这个组件中ID最小的那个任务。可以保证所有数据项只会被目标组件的一份实例(一个bolt)所处理

builder.setBolet(boltName,new BlackListBolt(),3).globalGrouping(SpoutName);

4.Fiellds Grouping 按域分组

builder.setBolet(boltName,new BlackListBolt(),3).fieldsGrouping(spoutName,new Field("域名");

源组件将其发送的数据项,按Tuple中指定域的值分组,向下游目标组件发送,可以保证拥有相同域组合的值的Tuple,被发送给同一个Bolt.

5、Direct Grouping 直接分组

builder.setSpout("kafkaSpout",topicSpout)
builder.setBolt(boltname1,new boltName1(),1).shuffleGrouping("kafkaSpout");
//以直接分组的模式接收上述bolt发送的数据项
builder.setBolt(boltname2,new boltname2(),2).directGrouping(boltname1);

源组件将其发送的数据项,以直接指定目标组件的方式发送,可以使指定组件接收给定的Tuple.需要注意的是,接收bolt的executle()函数中,哟啊使用emitDirect()替代emit,用于向指定的具名流中发送数据项

构建Topology

构建TopologyBuilder主要给出了三类方法:创建Topology、增加bolt和增加Spout的方法。setBolt和setSpout接口各有不同多种重载方法,均返回用于声明组件输入的对象。

1、id:组件(spout、Bolt)的标识,字符串类型,若需要引用该组件,就使用这里指定的标识ID。比如使用"kafkaSpout"

2、bolt:添加的bolt对象,再setBolt的重载方法中,存在IRichBolt和IBasicBolt两类bolt参数,项目中用到的是IRichBolt,区别在于,BasicBolt用于非聚集处理,能够自动进行(anchoring)和(acking)

3、spout:添加的Spout对象,在setSpout方法中该参数是IRichSpout类型的Spout接口。

4、parallelism_hint:并行度,数值型参数。设置组件运行时将要被分配的线程数量。

参考:《Storm 大数据流式计算及应用实践》

时间: 2024-10-12 21:23:31

Storm Topology及分组整理的相关文章

storm的流分组

用的是ShuffleGrouping分组方式,并行度设置为3 这是跑下来的结果 参考代码StormTopologyShufferGrouping.java package yehua.storm; import java.util.Map; import org.apache.storm.Config;import org.apache.storm.LocalCluster;import org.apache.storm.StormSubmitter;import org.apache.stor

第一个非常简单的storm topology的提交运行

配置好storm之后就可以开始在eclipse里面写topology了. 下面是我在网上搜到的一个简单的例子,我按照自己的理解注释了一下. 第一步,创建mvn工程 这是pom.xml文件 1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 2 xsi:schemaLocation="http

Caused by: java.lang.ClassNotFoundException: backtype.storm.topology.IRichSpout

1:初次运行Strom程序出现如下所示的错误,贴一下,方便脑补,也希望帮助到看到的小伙伴: 错误如下所示,主要问题是刚开始使用maven获取jar包的时候需要写<scope>provided</scope>,运行的时候需要把这行注释了即可,这是作用域的问题,开始需要在本地下载jar包,但是在虚拟机运行的时候已经存在这些jar包了,所以再写这句话就冲突了: 1 java.lang.NoClassDefFoundError: backtype/storm/topology/IRichS

Storm Topology Parallelism

Understanding the Parallelism of a Storm Topology What makes a running topology: worker processes, executors and tasks 在一个Strom集群中,实际运行一个topology有三个主要的实体 Worker processes Executors (threads) Tasks 下面是一张草图简单说明他们之间的关系: A worker process executes a subse

Storm Topology 提交到集群

问题:当完成Topology各个组件的定义之后(写好了**Spout.java 和 **Bolt.java)如何将Topology提交到集群中去? 参考:http://www.cnblogs.com/fxjwind/archive/2013/06/05/3119056.html 1,在**Topology.java中的main方法 setSpout.setBolt 之后通过TopologyBuilder.createTopology()创建Topology对象,在**Topology的main方

Storm系列之一——Storm Topology并发

1.是什么构成一个可运行的topology? worker processes(worker进程),executors(线程)和tasks. 一台Storm集群里面的机器可能运行一个或多个worker进程,一个worker进程运行一个特定topology的executors. 一个worker进程可能运行一个或多个executors.每个executor是一个线程.一个executor运行同一个spout或者bolt的一个或多个task. 一个task完成具体的数据处理. 一个worker进程执

Storm笔记整理(四):Storm核心概念与验证——并行度与流式分组

[TOC] Storm核心概念之并行度 Work 1个worker进程执行的是1个topology的子集(注:不会出现1个worker为多个topology服务).1个worker进程会启动1个或多个executor线程来执行1个topology的(spout或bolt).因此,1个运行中的topology就是由集群中多台(可能是一台)物理机上的一个或者多个worker进程组成的. Executor executor是worker进程启动的一个单独线程. 每个executor只会运行1个topo

Storm笔记整理(三):Storm集群安装部署与Topology作业提交

[TOC] Storm分布式集群安装部署 概述 Storm集群表面类似Hadoop集群.但在Hadoop上你运行的是"MapReduce jobs",在Storm上你运行的是"topologies"."Jobs"和"topologies"是大不同的,一个关键不同是一个MapReduce的Job最终会结束,而一个topology永远处理消息(或直到你kill它). Storm集群有两种节点:控制(master)节点和工作者(wo

Storm基本概念以及Topology的并发度

Spouts,流的源头 Spout是Storm里面特有的名词,Stream的源头,通常是从外部数据源读取tuples,并emit到topology Spout可以同时emit多个tupic stream,通过OutputFieldsDeclarer中的declareStream,method来定义 Spout需要实现RichSpout端口,最重要的方法是nextTuple,storm会不断调用接口从spout中取数据,同时需要注意的是Spout分为reliable or unreliable两种