`
农村外出务工男JAVA
  • 浏览: 105610 次
  • 性别: Icon_minigender_1
  • 来自: 重庆
社区版块
存档分类
最新评论

storm trident实战 分组聚合

阅读更多

一、前言

      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         
  • 大小: 8.9 KB
分享到:
评论

相关推荐

    Storm Trident实战之计算网站PV.rar

    本实战案例将重点介绍如何使用Storm Trident来计算网站的页面浏览量(PV,Page View)。 页面浏览量是衡量一个网站受欢迎程度的重要指标,通常通过记录用户对各个页面的访问次数来计算。在传统的批处理场景下,这...

    Storm Trident API 使用详解.docx

    4. Grouped Streams:在分组的 Stream 上进行的操作,如分组聚合,可以对特定字段进行分组后再进行聚合。 5. Merge 和 Join:合并(Merge)操作将多个 Stream 合并为一个,而 Join 操作则将来自不同 Stream 的数据...

    Storm入门教程 之Storm原理和概念详解

    Storm流计算从入门到精通之技术篇(高并发策略、批处理事务、Trident精解、运维监控、企业场景) Storm入门教程 之Storm原理和概念详解,出自Storm流计算从入门到精通之技术篇,Storm入门视频教程用到技术:Storm集群...

    Storm_Trident

    **Storm Trident:分布式流处理框架详解** Storm Trident是Twitter开源的、基于Apache Storm的一个高级抽象,它提供了一种更强大且高效的方式来处理实时数据流。Trident的核心理念是将数据流划分为一系列的小批量...

    Storm流计算之项目开发视频教程

    Storm视频教程通过含3个Storm完整项目,均为企业实际项目,其中一个是完全由Storm Trident开发。本课程每个技术均采用最新稳定版本,学完后可以从Kafka到Storm项目开发及HighCharts图表开发一个人搞定!涨工资?身价...

    基于Storm流计算天猫双十一作战室项目实战

    ### 基于Storm流计算天猫双十一作战室项目实战 #### 一、课程亮点与核心知识点 本课程针对Apache Storm这一强大的分布式实时计算系统进行了全方位的解析与实战演练。通过本课程的学习,不仅可以掌握Storm的基本...

    大数据分析架构师顶级培训课程storm课程 Trident理论与应用 Trident基础理论与实战 共35页.pptx

    ### 大数据开发高级就业指导课程——Storm及Trident理论与实战 #### 一、Storm并发机制 在Storm中,为了提高数据处理的性能和效率,设计了一套完整的并发机制。这一机制涉及到Topology的组件配置、并发度设置等多...

    Storm实战:构建大数据实时计算

     《Storm实战:构建大数据实时计算》是一本系统并且具有实践指导意义的Storm工具书和参考书,对Storm整个技术体系进行了全面的讲解,不仅包括对基本概念、特性的介绍,也涵盖了一些原理说明。  实战性很强,各章节...

    storm企业应用 实战 运维和调优

    但是根据标题和描述中的信息,我能够为你生成关于Storm企业应用实战运维和调优的知识点。 Apache Storm是一个实时计算的分布式计算框架,它类似于Hadoop,但它是为了实时处理而不是批处理设计的。Storm可以处理大量...

    trident-tutorial:实用的Storm Trident教程

    三叉戟教程实用的Storm Trident教程本教程以的的出色为基础。 流浪者的设置基于Taylor Goetz的。 Hazelcast状态代码基于wurstmeister的。 看看随附的。本教程的结构浏览Part * .java,了解Trident的基础知识使用...

    Storm 实战:构建大数据实时计算 PDF带书签完整版

    《Storm实战:构建大数据实时计算》是一本深入探讨Apache Storm技术的专业书籍,旨在帮助读者理解和掌握如何使用Storm进行大数据实时处理。Apache Storm是一个开源的分布式实时计算系统,它能够处理无限的数据流,...

    大数据处理框架:Storm:高级Storm:Trident与Stateful处理.docx

    大数据处理框架:Storm:高级Storm:Trident与Stateful处理.docx

    trident-examples:用 Storm Trident 编写的一组应用程序

    三叉戟的例子一组用 Storm Trident 编写的应用程序。应用用法建造$ git clone git@github.com:mayconbordin/trident-examples.git$ cd trident-examples/$ mvn -P &lt; profile&gt; package 使用local配置文件以本地模式...

    Storm实战构建大数据实时计算

    《Storm实战构建大数据实时计算》是一本专注于大数据领域实时处理技术的专著,主要围绕Apache Storm这一开源流处理系统展开。Storm被广泛应用于实时数据分析、在线机器学习、持续集成、实时网站仪表板等多个场景,其...

    Storm流计算项目:1号店电商实时数据分析系统-01.Storm项目实战课程大纲.pptx

    【课程大纲】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 storm trident自定义state的实现,实现了3种插入数据库的持久化方法,事务,不透明事务,不透明分区事务 下载下面,运行:mvn eclipse:eclipse即可在eclipse使用此项目。

    Storm实战构建大数据实时计算( 带书签目录 高清完整版)

    《Storm实战构建大数据实时计算》是一本专注于大数据处理领域的专著,主要围绕开源分布式实时计算系统Apache Storm展开。Apache Storm是一个强大的工具,用于处理大规模的数据流处理,它以高吞吐量、容错性以及实时...

    trident-elasticsearch:ElasticSearch 的 Storm Trident 集成层

    "trident-elasticsearch"项目是将这两者结合的产物,它提供了一个Storm Trident的集成层,使得在Storm中处理的数据能够无缝地流入和流出Elasticsearch。 首先,让我们深入了解一下Elasticsearch。Elasticsearch基于...

    storm-trident:《风暴蓝图》

    三叉戟《风暴蓝图:分布式实时计算模式》一书的源码和翻译=============(已完成,待校对)(未开始)(已完成,待校对)(已完成,待校对)(未开始)(未开始)(进行中)(未开始)(未开始)(未开始)

Global site tag (gtag.js) - Google Analytics