知识点
相关文章
更多最近更新
更多Storm常见模式——流聚合
2019-03-02 23:44|来源: 网路
流聚合(stream join)是指将具有共同元组(tuple)字段的数据流(两个或者多个)聚合形成一个新的数据流的过程。
从定义上看,流聚合和SQL中表的聚合(table join)很像,但是二者有明显的区别:table join的输入是有限的,并且join的语义是非常明确的;而流聚合的语义是不明确的并且输入流是无限的。
数据流的聚合类型跟具体的应用有关。一些应用把两个流发出的所有的tuple都聚合起来——不管多长时间;而另外一些应用则只会聚合一些特定的tuple。而另外一些应用的聚合逻辑又可能完全不一样。而这些聚合类型里面最常见的类型是把所有的输入流进行一样的划分,这个在storm里面用fields grouping在相同字段上进行grouping就可以实现。
下面是对storm-starter(代码见:https://github.com/nathanmarz/storm-starter)中有关两个流的聚合的示例代码剖析:
先看一下入口类SingleJoinExample。
(1)这里首先创建了两个发射源spout,分别是genderSpout和ageSpout:
FeederSpout genderSpout = new FeederSpout(new Fields("id", "gender")); FeederSpout ageSpout = new FeederSpout(new Fields("id", "age")); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("gender", genderSpout); builder.setSpout("age", ageSpout);
其中genderSpout包含两个tuple字段:id和gender,ageSpout包含两个tuple字段:id和age(这里流聚合就是通过将相同id的tuple进行聚合,得到一个新的输出流,包含id、gender和age字段)。
(2)为了不同的数据流中的同一个id的tuple能够落到同一个task中进行处理,这里使用了storm中的fileds grouping在id字段上进行分组划分:
builder.setBolt("join", new SingleJoinBolt(new Fields("gender", "age"))) .fieldsGrouping("gender", new Fields("id")) .fieldsGrouping("age", new Fields("id"));
从中可以看到,SingleJoinBolt就是真正进行流聚合的地方。下面我们来看看:
(1)SingleJoinBolt构造时接收一个Fileds对象,其中传进的是聚合后将要被输出的字段(这里就是gender和age字段),保存到变量_outFileds中。
(2)接下来看看完成SingleJoinBolt的构造后,SingleJoinBolt在真正开始接收处理tuple之前所做的准备工作(代码见prepare方法):
a)首先,将保存OutputCollector对象,创建TimeCacheMap对象,设置超时回调接口,用于tuple处理失败时fail消息;紧接着记录数据源的个数:
_collector = collector; int timeout = ((Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)).intValue(); _pending = new TimeCacheMap<List<Object>, Map<GlobalStreamId, Tuple>>(timeout, new ExpireCallback()); _numSources = context.getThisSources().size();
b)遍历TopologyContext中不同数据源,得到所有数据源(这里就是genderSpout和ageSpout)中公共的Filed字段,保存到变量_idFields中(例子中就是id字段),同时将_outFileds中字段所在数据源记录下来,保存到一张HashMap中_fieldLocations,以便聚合后获取对应的字段值。
Set<String> idFields = null; for(GlobalStreamId source: context.getThisSources().keySet()) { Fields fields = context.getComponentOutputFields(source.get_componentId(), source.get_streamId()); Set<String> setFields = new HashSet<String>(fields.toList()); if(idFields==null) idFields = setFields; else idFields.retainAll(setFields); for(String outfield: _outFields) { for(String sourcefield: fields) { if(outfield.equals(sourcefield)) { _fieldLocations.put(outfield, source); } } } } _idFields = new Fields(new ArrayList<String>(idFields)); if(_fieldLocations.size()!=_outFields.size()) { throw new RuntimeException("Cannot find all outfields among sources"); }
(3)好了,下面开始两个spout流的聚合过程了(代码见execute方法):
首先,从tuple中获取_idFields字段,如果不存在于等待被处理的队列_pending中,则加入一行,其中key是获取到的_idFields字段,value是一个空的HashMap<GlobalStreamId, Tuple>对象,记录GlobalStreamId到Tuple的映射。
List<Object> id = tuple.select(_idFields); GlobalStreamId streamId = new GlobalStreamId(tuple.getSourceComponent(), tuple.getSourceStreamId()); if(!_pending.containsKey(id)) { _pending.put(id, new HashMap<GlobalStreamId, Tuple>()); }
从_pending队列中,获取当前GlobalStreamId streamId对应的HashMap对象parts中:
Map<GlobalStreamId, Tuple> parts = _pending.get(id);
如果streamId已经包含其中,则抛出异常,接收到同一个spout中的两条一样id的tuple,否则将该streamid加入parts中:
if(parts.containsKey(streamId)) throw new RuntimeException("Received same side of single join twice"); parts.put(streamId, tuple);
如果parts已经包含了聚合数据源的个数_numSources时,从_pending队列中移除这条记录,然后开始构造聚合后的结果字段:依次遍历_outFields中各个字段,从_fieldLocations中取到这些outFiled字段对应的GlobalStreamId,紧接着从parts中取出GlobalStreamId对应的outFiled,放入聚合后的结果中。
if(parts.size()==_numSources) { _pending.remove(id); List<Object> joinResult = new ArrayList<Object>(); for(String outField: _outFields) { GlobalStreamId loc = _fieldLocations.get(outField); joinResult.add(parts.get(loc).getValueByField(outField)); }
最后通过_collector将parts中存放的tuple和聚合后的输出结果发射出去,并ack这些tuple已经处理成功。
_collector.emit(new ArrayList<Tuple>(parts.values()), joinResult); for(Tuple part: parts.values()) { _collector.ack(part); }
}
否则,继续等待两个spout流中这个streamid都到齐后再进行聚合处理。
(4)最后,声明一下输出字段(代码见declareOutputFields方法):
declarer.declare(_outFields);
转自:http://www.cnblogs.com/panfeng412/archive/2012/06/04/storm-common-patterns-of-stream-join
相关问答
更多-
什么是常见的培训模式[2022-03-18]
-
storm本地模式是不是本地程序直接可以跑起来?[2022-03-13]
1.jpg 支持的,0.9开始支持得好一些在以前的版本里,如果想在Windows平台上运行Storm,你需要安装ZeroMQ,修改Storm的源码,追加一些Windows平台特定的脚本。而在新的版本里,因为用netty替换了ZeroMQ,由于netty用纯java实现,因此使得Storm具有更好的跨平台特性,现在要在Windows上运行Storm比以前容易很多。 -
linux的常见的分区类型有哪些?[2022-05-23]
ext2,ext3比较常用,但是最新的ubuntu己经在用ext4了。我用的就是。另个,还有一些嵌入式linux系统常用jffs和yaffs文件系统。而且,linux支持几乎所有你所知道的文件系统类型(也即分类类型).包括windows下常用的fat32 ,ntfs等。 -
java中几种常见的设计模式[2022-05-10]
一共23种设计模式! 按照目的来分,设计模式可以分为创建型模式、结构型模式和行为型模式。 创建型模式用来处理对象的创建过程;结构型模式用来处理类或者对象的组合;行为型模式用来对类或对象怎样交互和怎样分配职责进行描述。 创建型模式用来处理对象的创建过程,主要包含以下5种设计模式: 工厂方法模式(Factory Method Pattern) 抽象工厂模式(Abstract Factory Pattern) 建造者模式(Builder Pattern) 原型模式(Prototype Patter ... -
JAVA中常见的设计模式都有啥[2023-11-15]
http://zhidao.baidu.com/question/20013947.html?fr=qrl3 -
web前端常见的代码设计模式有哪些[2023-02-04]
常见代码 function CreatePerson(name,age,sex) { var obj = new Object(); obj.name = name; obj.age = age; obj.sex = sex; obj.sayName = function(){ return this.name; } return obj; } var p1 = new CreatePerson("longen",'28','男'); var p2 = new CreatePerson("tugenhua" ... -
支架式教学设计的常见模式有哪些[2021-12-30]
问题支架 。问题是学习过程中最为常见的支架,相对“框架问题”而言,支架问题的系统性较弱,有经验的教师会在学生的学习过程中自然地、应机地提供此类支架。同时,在特定主题的学习中,“支架问题”往往比“框架问题”更具结构性,更加关注细节与可操作性 -
常见的教学模式有那些?[2023-04-22]
师生系统的传授和学习书本知识 教师辅导学生从活动中自己学习 折衷于俩者之间的教学模式 -
这是一个Spark(2.1.0)UDAF来计算给定列的统计模式: package org.anish.spark.mostcommonvalue import org.apache.spark.sql.Row import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} import org.apache.spark.sql.types._ import scalaz.Sc ...