Storm杂记 — Field Grouping和Shuffle Grouping的区别

转载请注明出处:http://blog.csdn.net/luonanqin

最近研究Storm的Stream Grouping的时候,对Field Grouping和Shuffle Grouping理解不是很透彻。去看WordCountTopology也不怎么理解,后来脑洞一开,加了一行代码再次运行,彻底顿悟。只能说自己对Storm的基本概念还是没吃透啊。(WordCountTopology这个例子请自行参考Storm-Starter)

public void execute(Tuple tuple, BasicOutputCollector collector) {
	String word = tuple.getString(0);

	// 添加这行代码的作用是看看值相等的word是不是同一个实例执行的,实时证明确实如此
	System.out.println(this + "====" + word);

	Integer count = counts.get(word);
	if (count == null)
		count = 0;
	count++;
	counts.put(word, count);
	collector.emit(new Values(word, count));
}

经过反复测试,下面是我个人的一些总结,如果有缺少或者错误我会及时改正。

官方文档里有这么一句话:“if the stream is grouped by the “user-id” field, tuples with the same “user-id” will always go to the same task”

一个task就是一个处理逻辑的实例,所以fields能根据tuple stream的id,也就是下面定义的xxx

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

xxx所代表的具体内容会由某一个task来处理,并且同一个xxx对应的内容,处理这个内容的task实例是同一个。

比如说:

bolt第一次emit三个流,即xxx有luonq pangyang qinnl三个值,假设分别建立三个task实例来处理:

luonq -> instance1
pangyang -> instance2
qinnl -> instance3

然后第二次emit四个流,即xxx有luonq qinnanluo py pangyang四个值,假设还是由刚才的三个task实例来处理:

luonq -> instance1
qinnanluo -> instance2
py -> instance3
pangyang -> instance2

然后第三次emit两个流,即xxx有py qinnl两个值,假设还是由刚才的三个task实例来处理:

py -> instance3
qinnl -> instance3

最后我们看看三个task实例都处理了哪些值,分别处理了多少次:

instance1: luonq(处理2次)

instance2: pangyang(处理2次) qinnanluo(处理1次)

instance3: qinnl(处理2次) py(处理2次)

结论:

1. emit发出的值第一次由哪个task实例处理是随机的,此后再次出现这个值,就固定由最初处理他的那个task实例再次处理,直到topology结束

2. 一个task实例可以处理多个emit发出的值

3. 和shuffle Grouping的区别就在于,当emit发出同样的值时,处理他的task是随机的

时间: 2024-10-13 14:31:31

Storm杂记 — Field Grouping和Shuffle Grouping的区别的相关文章

【转】rollup、cub、grouping sets、grouping、grouping_id在报表中的应用

摘自 http://blog.itpub.net/26977915/viewspace-734114/ 在报表语句中经常要使用各种分组汇总,rollup和cube就是常用的分组汇总方式. 第一:group by rollup 1.如果使用诸如group by rollup(A,B,C)的方式分组,那么返回的分组结果是(A,B,C) (A,B) (A) (NULL) 一共四种结果.即从右到左递减,最后来个合计. 例如: SQL> select * from t; YEARS     MONTHS

[转]hadoop,spark,storm,pig,hive,mahout等到底有什么区别和联系?

摘自知乎大神的论述 作者:Xiaoyu Ma链接:https://www.zhihu.com/question/27974418/answer/38965760来源:知乎著作权归作者所有,转载请联系作者获得授权. 大数据本身是个很宽泛的概念,Hadoop生态圈(或者泛生态圈)基本上都是为了处理超过单机尺度的数据处理而诞生的.你可以把它比作一个厨房所以需要的各种工具.锅碗瓢盆,各有各的用处,互相之间又有重合.你可以用汤锅直接当碗吃饭喝汤,你可以用小刀或者刨子去皮.但是每个工具有自己的特性,虽然奇怪

MR的shuffle和Spark的shuffle之间的区别

mr的shuffle mapShuffle 数据存到hdfs中是以块进行存储的,每一个块对应一个分片,maptask就是从分片中获取数据的 在某个节点上启动了map Task,map Task读取是通过k-v来读取的,读取的数据会放到环形缓存区,这样做的目的是为了防止IO的访问次数,然后环形缓存区的内存达到一定的阀值的 时候会把文件益写到磁盘,溢出的各种小文件会合并成一个大文件,这个合并的过程中会进行排序,这个排序叫做归并排序 map阶段会涉及到 1.sort排序(默认按字典排序) 2.合并(c

C#面向对象字段(Field)和属性(Property)的区别

C#中字段可以是公有或私有的,可以随意的改变字段的值,但是当需要限定字段的取值范围或者设置字段只读或只写,单靠字段是没办法完成的,这时就引入了属性,可以通过属性的get和set设置字段的读取权限,get模块负责属性的读取,set模块负责属性的写入 例:通过set限制字段的取值 public class Person { private int _age;//字段 public int Age//属性 { get { return _age; } set//可以限制字段的取值范围 { if (va

ThinkPHP中field 方法与getField 方法的区别。

做项目的时候遇到了一个问题,框架生成的sql与放到navicat中执行.和页面显示出来的结果不太一样. 排查了很久,也没有找到问题. 出现问题的sql如下. $fuck = M(null, null, 'DB_CONFIG1')->table("$tableName as p") ->join('left join 表一 as c on p.PL_CID = c.ID') ->join('left join 表二 as cl on p.PL_CLID = cl.CLI

关于Storm Stream grouping

在Storm中, 开发者可以为上游spout/bolt发射出的tuples指定下游bolt的哪个/哪些task(s)来处理该tuples.这种指定在storm中叫做对stream的分组,即stream grouping,分组方式主要有以下7种 Shuffle Grouping 或 None Grouping Fields Grouping All Grouping Global Grouping LocalOrShuffle Grouping Direct Grouping 1. Shuffle

stream grouping

stream grouping就是用来定义一个stream应该如果分配给Bolts上面的多个Tasks.   storm里面有6种类型的stream grouping: 1. Shuffle Grouping: 随机分组, 随机派发stream里面的tuple, 保证每个bolt接收到的tuple数目相同.轮询,平均分配. 2. Fields Grouping:按字段分组, 比如按userid来分组, 具有同样userid的tuple会被分到相同的Bolts, 而不同的userid则会被分配到不

storm - 简介

一 Storm简介 Storm是Twitter开源的一个类似于Hadoop的实时数据处理框架,它原来是由BackType开发,后BackType被Twitter收购,将Storm作为Twitter的实时数据分析系统. 实时数据处理的应用场景很广泛,例如商品推荐,广告投放,它能根据当前情景上下文(用户偏好,地理位置,已发生的查询和点击等)来估计用户点击的可能性并实时做出调整. twitter列举了storm的三大作用领域: 1.信息流处理(Stream Processing) Storm可以用来实

storm 原理简介及单机版安装指南(转)

本文翻译自: https://github.com/nathanmarz/storm/wiki/Tutorial Storm是一个分布式的.高容错的实时计算系统. Storm对于实时计算的的意义相当于Hadoop对于批处理的意义.Hadoop为我们提供了Map和Reduce原语,使我们对数据进行批处理变的非常的简单和优美.同样,Storm也对数据的实时计算提供了简单Spout和Bolt原语. Storm适用的场景: 1.流数据处理:Storm可以用来用来处理源源不断的消息,并将处理之后的结果保存