一、前言
groupBy分组操作,根据指定属性进行分组,如果后面是aggregate()的话,先根据partitionBy分区,在每个partition上分组,分完组后,在每个分组上进行聚合。
二、实战
main:
public static void main(String[] args) throws Exception {
/**
* 多设置几个并行度,分组后如果分组不够,那么将有并行度空闲跑者
*/
@SuppressWarnings("unchecked")
FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
new Values("a"), new Values("b"), new Values("a"),new Values("c"),
new Values("c"),new Values("c"),new Values("d"));
spout.setCycle(false);
TridentTopology tridentTopology = new TridentTopology();
tridentTopology
.newStream("spout", spout)
.parallelismHint(3)
.shuffle()
.groupBy(new Fields("sentence"))
.aggregate(new Fields("sentence"), new MyAgg(),
new Fields("Map"))
.parallelismHint(5)
.each(new Fields("sentence","Map"), new MyBolt());
Config config = new Config();
config.setDebug(false);
StormSubmitter.submitTopology("trident_groupby_aggregate_many", config,
tridentTopology.build());
}
MyAgg:
package com.storm.trident.groupby.先分组后聚合; import java.util.HashMap; import java.util.Map; import org.apache.storm.trident.operation.BaseAggregator; import org.apache.storm.trident.operation.TridentCollector; import org.apache.storm.trident.operation.TridentOperationContext; import org.apache.storm.trident.tuple.TridentTuple; import org.apache.storm.tuple.Values; public class MyAgg extends BaseAggregator<Map<String, Integer>> { private static final long serialVersionUID = 1L; /** * 属于哪个分区 */ private int partitionId; /** * 分区数量 */ private int numPartitions; @SuppressWarnings("rawtypes") @Override public void prepare(Map conf, TridentOperationContext context) { partitionId = context.getPartitionIndex(); numPartitions = context.numPartitions(); } @Override public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) { String word = tuple.getString(0); Integer value = val.get(word); if (value == null) { value = 0; } value++; // 把数据保存到一个map对象中 val.put(word, value); val.put(word + "属于哪个分区", partitionId); System.out.println("I am partition [" + partitionId + "] and I have kept a tweet by: " + numPartitions); } @Override public void complete(Map<String, Integer> val, TridentCollector collector) { collector.emit(new Values(val)); } @Override public Map<String, Integer> init(Object arg0, TridentCollector arg1) { return new HashMap<String, Integer>(); } }
MyBolt:
package com.storm.trident.groupby.先分组后聚合; import java.util.Map; import java.util.Map.Entry; import org.apache.storm.trident.operation.BaseFilter; import org.apache.storm.trident.tuple.TridentTuple; public class MyBolt extends BaseFilter { /** * */ private static final long serialVersionUID = 1L; @SuppressWarnings("unchecked") @Override public boolean isKeep(TridentTuple tuple) { System.out.println("打印出来的tuple:" + tuple); Map<String, Integer> value = ((Map<String,Integer>) tuple.getValue(1)); for (Entry<String, Integer> entry : value.entrySet()) { System.out.println("key:"+ entry.getKey()+",value:" + entry.getValue()); } return false; } }
三、测试
打包在storm集群里跑
查看log日志,主要日志如下
2016-12-22 18:36:11.293 STDIO [INFO] I am partition [3] and I have kept a tweet by: 5 2016-12-22 18:36:11.302 STDIO [INFO] I am partition [4] and I have kept a tweet by: 5 2016-12-22 18:36:11.304 STDIO [INFO] 打印出来的tuple:[b, {b属于哪个分区=4, b=1}] 2016-12-22 18:36:11.306 STDIO [INFO] key:b属于哪个分区,value:4 2016-12-22 18:36:11.317 STDIO [INFO] I am partition [3] and I have kept a tweet by: 5 2016-12-22 18:36:11.321 STDIO [INFO] key:b,value:1 2016-12-22 18:36:11.335 STDIO [INFO] 打印出来的tuple:[a, {a属于哪个分区=3, a=2}] 2016-12-22 18:36:11.341 STDIO [INFO] key:a属于哪个分区,value:3 2016-12-22 18:36:11.344 STDIO [INFO] key:a,value:2 2016-12-22 18:36:11.423 STDIO [INFO] I am partition [0] and I have kept a tweet by: 5 2016-12-22 18:36:11.424 STDIO [INFO] I am partition [0] and I have kept a tweet by: 5 2016-12-22 18:36:11.425 STDIO [INFO] I am partition [0] and I have kept a tweet by: 5 2016-12-22 18:36:11.433 STDIO [INFO] 打印出来的tuple:[c, {c=3, c属于哪个分区=0}] 2016-12-22 18:36:11.433 STDIO [INFO] key:c,value:3 2016-12-22 18:36:11.439 STDIO [INFO] key:c属于哪个分区,value:0 2016-12-22 18:36:11.923 STDIO [INFO] I am partition [1] and I have kept a tweet by: 5 2016-12-22 18:36:11.933 STDIO [INFO] 打印出来的tuple:[d, {d=1, d属于哪个分区=1}] 2016-12-22 18:36:11.939 STDIO [INFO] key:d,value:1 2016-12-22 18:36:11.942 STDIO [INFO] key:d属于哪个分区,value:1
相关推荐
本实战案例将重点介绍如何使用Storm Trident来计算网站的页面浏览量(PV,Page View)。 页面浏览量是衡量一个网站受欢迎程度的重要指标,通常通过记录用户对各个页面的访问次数来计算。在传统的批处理场景下,这...
4. Grouped Streams:在分组的 Stream 上进行的操作,如分组聚合,可以对特定字段进行分组后再进行聚合。 5. Merge 和 Join:合并(Merge)操作将多个 Stream 合并为一个,而 Join 操作则将来自不同 Stream 的数据...
Storm流计算从入门到精通之技术篇(高并发策略、批处理事务、Trident精解、运维监控、企业场景) Storm入门教程 之Storm原理和概念详解,出自Storm流计算从入门到精通之技术篇,Storm入门视频教程用到技术:Storm集群...
**Storm Trident:分布式流处理框架详解** Storm Trident是Twitter开源的、基于Apache Storm的一个高级抽象,它提供了一种更强大且高效的方式来处理实时数据流。Trident的核心理念是将数据流划分为一系列的小批量...
Storm视频教程通过含3个Storm完整项目,均为企业实际项目,其中一个是完全由Storm Trident开发。本课程每个技术均采用最新稳定版本,学完后可以从Kafka到Storm项目开发及HighCharts图表开发一个人搞定!涨工资?身价...
### 基于Storm流计算天猫双十一作战室项目实战 #### 一、课程亮点与核心知识点 本课程针对Apache Storm这一强大的分布式实时计算系统进行了全方位的解析与实战演练。通过本课程的学习,不仅可以掌握Storm的基本...
### 大数据开发高级就业指导课程——Storm及Trident理论与实战 #### 一、Storm并发机制 在Storm中,为了提高数据处理的性能和效率,设计了一套完整的并发机制。这一机制涉及到Topology的组件配置、并发度设置等多...
《Storm实战:构建大数据实时计算》是一本系统并且具有实践指导意义的Storm工具书和参考书,对Storm整个技术体系进行了全面的讲解,不仅包括对基本概念、特性的介绍,也涵盖了一些原理说明。 实战性很强,各章节...
但是根据标题和描述中的信息,我能够为你生成关于Storm企业应用实战运维和调优的知识点。 Apache Storm是一个实时计算的分布式计算框架,它类似于Hadoop,但它是为了实时处理而不是批处理设计的。Storm可以处理大量...
三叉戟教程实用的Storm Trident教程本教程以的的出色为基础。 流浪者的设置基于Taylor Goetz的。 Hazelcast状态代码基于wurstmeister的。 看看随附的。本教程的结构浏览Part * .java,了解Trident的基础知识使用...
《Storm实战:构建大数据实时计算》是一本深入探讨Apache Storm技术的专业书籍,旨在帮助读者理解和掌握如何使用Storm进行大数据实时处理。Apache Storm是一个开源的分布式实时计算系统,它能够处理无限的数据流,...
大数据处理框架:Storm:高级Storm:Trident与Stateful处理.docx
三叉戟的例子一组用 Storm Trident 编写的应用程序。应用用法建造$ git clone git@github.com:mayconbordin/trident-examples.git$ cd trident-examples/$ mvn -P < profile> package 使用local配置文件以本地模式...
《Storm实战构建大数据实时计算》是一本专注于大数据领域实时处理技术的专著,主要围绕Apache Storm这一开源流处理系统展开。Storm被广泛应用于实时数据分析、在线机器学习、持续集成、实时网站仪表板等多个场景,其...
【课程大纲】01.Storm项目实战课程大纲02.CDH5搭建之CM5安装部署03.CDH5搭建和CM界面化集群管理04.Hadoop、HBase、Zookeeper集群管理和角色分配05.Kafka基础知识和集群搭建06.Kafka基本操作和最优设置07.Kafka Java ...
storm_trident_state storm trident自定义state的实现,实现了3种插入数据库的持久化方法,事务,不透明事务,不透明分区事务 下载下面,运行:mvn eclipse:eclipse即可在eclipse使用此项目。
《Storm实战构建大数据实时计算》是一本专注于大数据处理领域的专著,主要围绕开源分布式实时计算系统Apache Storm展开。Apache Storm是一个强大的工具,用于处理大规模的数据流处理,它以高吞吐量、容错性以及实时...
"trident-elasticsearch"项目是将这两者结合的产物,它提供了一个Storm Trident的集成层,使得在Storm中处理的数据能够无缝地流入和流出Elasticsearch。 首先,让我们深入了解一下Elasticsearch。Elasticsearch基于...
三叉戟《风暴蓝图:分布式实时计算模式》一书的源码和翻译=============(已完成,待校对)(未开始)(已完成,待校对)(已完成,待校对)(未开始)(未开始)(进行中)(未开始)(未开始)(未开始)