【flink番外篇】1、flink的23种常用算子介绍及详细示例（2）- keyby、reduce和Aggregations

Flink 系列文章

1、Flink 专栏等系列综合文章链接

文章目录

  • Flink 系列文章
  • 一、Flink的23种算子说明及示例
    • 6、KeyBy
    • 7、Reduce
    • 8、Aggregations

本文主要介绍Flink 的3种常用的operator（keyby、reduce和Aggregations）及以具体可运行示例进行说明.
如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。
本文除了maven依赖外，没有其他依赖。

本专题分为五篇，即：
【flink番外篇】1、flink的23种常用算子介绍及详细示例（1）- map、flatmap和filter
【flink番外篇】1、flink的23种常用算子介绍及详细示例（2）- keyby、reduce和Aggregations
【flink番外篇】1、flink的23种常用算子介绍及详细示例（3）-window、distinct、join等
【flink番外篇】1、flink的23种常用算子介绍及详细示例（4）- union、window join、connect、outputtag、cache、iterator、project
【flink番外篇】1、flink的23种常用算子介绍及详细示例（完整版）

一、Flink的23种算子说明及示例

本文示例中使用的maven依赖和java bean 参考本专题的第一篇中的maven和java bean。

6、KeyBy

DataStream → KeyedStream
按照指定的key来对流中的数据进行分组
在这里插入图片描述
KeyBy 在逻辑上是基于 key 对流进行分区。在内部，它使用 hash 函数对流进行分区。它返回 KeyedDataStream 数据流。将同一Key的数据放到同一个分区。

分区结果和KeyBy下游算子的并行度强相关。如下游算子只有一个并行度，不管怎么分，都会分到一起。
对于POJO类型，KeyBy可以通过keyBy(fieldName)指定字段进行分区。
对于Tuple类型，KeyBy可以通过keyBy(fieldPosition)指定字段进行分区。
对于一般类型，如上，KeyBy可以通过keyBy(new KeySelector {…})指定字段进行分区。

import java.util.Arrays;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.datastreamapi.User;/*** @author alanchan**/
public class TestKeyByDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		env.setParallelism(4);// 设置数据分区数量keyByFunction6(env);env.execute();}// 构造User数据源public static DataStreamSource<User> source(StreamExecutionEnvironment env) {DataStreamSource<User> source = env.fromCollection(Arrays.asList(new User(1, "alan1", "1", "1@1.com", 12, 1000), new User(2, "alan2", "2", "2@2.com", 19, 200),new User(3, "alan1", "3", "3@3.com", 28, 1500), new User(5, "alan1", "5", "5@5.com", 15, 500), new User(4, "alan2", "4", "4@4.com", 30, 400)));return source;}// 按照name进行keyby 对于POJO类型，KeyBy可以通过keyBy(fieldName)指定字段进行分区public static void keyByFunction1(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> sink = source.keyBy(new KeySelector<User, String>() {@Overridepublic String getKey(User value) throws Exception {return value.getName();}});sink.map(user -> {System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + user.toString());return user;});sink.print();}// lambda 对于POJO类型，KeyBy可以通过keyBy(fieldName)指定字段进行分区public static void keyByFunction2(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> sink = source.keyBy(user -> user.getName());// 演示keyby后的数据输出sink.map(user -> {System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + user.toString());return user;});sink.print();}// 对于Tuple类型，KeyBy可以通过keyBy(fieldPosition)指定字段进行分区。lambdapublic static void keyByFunction3(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);SingleOutputStreamOperator<Tuple2<String, User>> userTemp = source.map((MapFunction<User, Tuple2<String, User>>) user -> {return new Tuple2<String, User>(user.getName(), user);}).returns(Types.TUPLE(Types.STRING, Types.POJO(User.class)));KeyedStream<Tuple2<String, User>, Tuple> sink = userTemp.keyBy(0);// 演示keyby后的数据输出sink.map(user -> {System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + user.f1.toString());return user.f1;});sink.print();}// 对于Tuple类型，KeyBy可以通过keyBy(fieldPosition)指定字段进行分区。public static void keyByFunction4(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);SingleOutputStreamOperator<Tuple2<String, User>> userTemp = source.map(new MapFunction<User, Tuple2<String, User>>() {@Overridepublic Tuple2<String, User> map(User value) throws Exception {return new Tuple2<String, User>(value.getName(), value);}});KeyedStream<Tuple2<String, User>, String> sink = userTemp.keyBy(new KeySelector<Tuple2<String, User>, String>() {@Overridepublic String getKey(Tuple2<String, User> value) throws Exception {return value.f0;}});// 演示keyby后的数据输出sink.map(user -> {System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + user.f1.toString());return user.f1;});//		sink.map(new MapFunction<Tuple2<String, User>, String>() {
//
//			@Override
//			public String map(Tuple2<String, User> value) throws Exception {
//				System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + value.f1.toString());
//				return null;
//			}});sink.print();}// 对于一般类型，如上，KeyBy可以通过keyBy(new KeySelector {...})指定字段进行分区。// 按照name的前4位进行keybypublic static void keyByFunction5(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> sink = source.keyBy(new KeySelector<User, String>() {@Overridepublic String getKey(User value) throws Exception {
//				String temp = value.getName().substring(0, 4);return value.getName().substring(0, 4);}});sink.map(user -> {System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + user.toString());return user;});sink.print();}// 对于一般类型，如上，KeyBy可以通过keyBy(new KeySelector {...})指定字段进行分区。 lambda// 按照name的前4位进行keybypublic static void keyByFunction6(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> sink = source.keyBy(user -> user.getName().substring(0, 4));sink.map(user -> {System.out.println("当前线程ID：" + Thread.currentThread().getId() + ",user:" + user.toString());return user;});sink.print();}}

7、Reduce

KeyedStream → DataStream
对集合中的元素进行聚合。Reduce 返回单个的结果值，并且 reduce 操作每处理一个元素总是创建一个新值。常用的方法有 average, sum, min, max, count，使用 reduce 方法都可实现。基于ReduceFunction进行滚动聚合，并向下游算子输出每次滚动聚合后的结果。
注意: Reduce会输出每一次滚动聚合的结果。
在这里插入图片描述

import java.util.Arrays;import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.datastreamapi.User;/*** @author alanchan**/
public class TestReduceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		env.setParallelism(4);// 设置数据分区数量reduceFunction2(env);env.execute();}// 构造User数据源public static DataStreamSource<User> source(StreamExecutionEnvironment env) {DataStreamSource<User> source = env.fromCollection(Arrays.asList(new User(1, "alan1", "1", "1@1.com", 12, 1000), new User(2, "alan2", "2", "2@2.com", 19, 200),new User(3, "alan1", "3", "3@3.com", 28, 1500), new User(5, "alan1", "5", "5@5.com", 15, 500), new User(4, "alan2", "4", "4@4.com", 30, 400)));return source;}// 按照name进行balance进行sumpublic static void reduceFunction1(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> keyedStream = source.keyBy(user -> user.getName());SingleOutputStreamOperator<User> sink = keyedStream.reduce(new ReduceFunction<User>() {@Overridepublic User reduce(User value1, User value2) throws Exception {double balance = value1.getBalance() + value2.getBalance();return new User(value1.getId(), value1.getName(), "", "", 0, balance);}});//sink.print();}// 按照name进行balance进行sum lambdapublic static void reduceFunction2(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> userKeyBy = source.keyBy(user -> user.getName());SingleOutputStreamOperator<User> sink = userKeyBy.reduce((user1, user2) -> {User user = user1;user.setBalance(user1.getBalance() + user2.getBalance());return user;});sink.print();}}

8、Aggregations

KeyedStream → DataStream
DataStream API 支持各种聚合，例如 min，max，sum 等。 这些函数可以应用于 KeyedStream 以获得 Aggregations 聚合。
Aggregate 对KeyedStream按指定字段滚动聚合并输出每一次滚动聚合后的结果。默认的聚合函数有:sum、min、minBy、max、maxBy。
注意:
max(field)与maxBy(field)的区别: maxBy返回field最大的那条数据;而max则是将最大的field的值赋值给第一条数据并返回第一条数据。同理,min与minBy。
Aggregate聚合算子会滚动输出每一次聚合后的结果
max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键， min 和 minBy 同理。
max以第一个比较对象的比较列值进行替换，maxBy是以整个比较对象进行替换。


import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.datastreamapi.User;/*** @author alanchan**/
public class TestAggregationsDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();aggregationsFunction2(env);env.execute();}// 构造User数据源public static DataStreamSource<User> source(StreamExecutionEnvironment env) {DataStreamSource<User> source = env.fromCollection(Arrays.asList(new User(1, "alan1", "1", "1@1.com", 12, 1000), new User(2, "alan2", "2", "2@2.com", 19, 200),new User(3, "alan1", "3", "3@3.com", 28, 1500), new User(5, "alan1", "5", "5@5.com", 15, 500), new User(4, "alan2", "4", "4@4.com", 30, 400)));return source;}//分组统计sum、max、min、maxby、minbypublic static void aggregationsFunction(StreamExecutionEnvironment env) throws Exception {DataStreamSource<User> source = source(env);KeyedStream<User, String> userTemp=	source.keyBy(user->user.getName());DataStream sink = null;//1、根据name进行分区统计balance之和 alan1----2500/alan2----600
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=2500.0)
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=600.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=3000.0)sink = userTemp.sum("balance");//2、根据name进行分区统计balance的max alan1----1500/alan2----400
//		 1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)
//		 16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		 16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1500.0)
//		 1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=400.0)
//		 16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1500.0)sink = userTemp.max("balance");//1@1.com-3000 --  2@2.com-300//3、根据name进行分区统计balance的min  alan1----500/alan2---200
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=500.0)
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)sink = userTemp.min("balance");//4、根据name进行分区统计balance的maxBy alan2----400/alan1----1500
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)
//		1> User(id=4, name=alan2, pwd=4, email=4@4.com, age=30, balance=400.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		16> User(id=3, name=alan1, pwd=3, email=3@3.com, age=28, balance=1500.0)
//		16> User(id=3, name=alan1, pwd=3, email=3@3.com, age=28, balance=1500.0)sink = userTemp.maxBy("balance");//5、根据name进行分区统计balance的minBy alan2----200/alan1----500
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)
//		1> User(id=2, name=alan2, pwd=2, email=2@2.com, age=19, balance=200.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		16> User(id=1, name=alan1, pwd=1, email=1@1.com, age=12, balance=1000.0)
//		16> User(id=5, name=alan1, pwd=5, email=5@5.com, age=15, balance=500.0)sink = userTemp.minBy("balance");sink.print();}public static void aggregationsFunction2(StreamExecutionEnvironment env) throws Exception {List list = new ArrayList<Tuple3<Integer, Integer, Integer>>();list.add(new Tuple3<>(0,3,6));list.add(new Tuple3<>(0,2,5));list.add(new Tuple3<>(0,1,6));list.add(new Tuple3<>(0,4,3));list.add(new Tuple3<>(1,1,9));list.add(new Tuple3<>(1,2,8));list.add(new Tuple3<>(1,3,10));list.add(new Tuple3<>(1,2,9));list.add(new Tuple3<>(1,5,7));DataStreamSource<Tuple3<Integer, Integer, Integer>> source = env.fromCollection(list);KeyedStream<Tuple3<Integer, Integer, Integer>, Integer> tTemp=  source.keyBy(t->t.f0);DataStream<Tuple3<Integer, Integer, Integer>> sink =null;//按照分区，以第一个Tuple3的元素为基础进行第三列值比较，如果第三列值小于第一个tuple3的第三列值，则进行第三列值替换，其他的不变
//        12> (0,3,6)
//        11> (1,1,9)
//        11> (1,1,8)
//        12> (0,3,5)
//        11> (1,1,8)
//        12> (0,3,5)
//        11> (1,1,8)
//        12> (0,3,3)
//        11> (1,1,7)  sink =  tTemp.min(2);//     按照数据分区，以第一个tuple3的元素为基础进行第三列值比较，如果第三列值小于第一个tuple3的第三列值，则进行整个tuple3的替换
//     12> (0,3,6)
//     11> (1,1,9)
//     12> (0,2,5)
//     11> (1,2,8)
//     12> (0,2,5)
//     11> (1,2,8)
//     12> (0,4,3)
//     11> (1,2,8)
//     11> (1,5,7)sink = tTemp.minBy(2);sink.print();}}

以上，本文主要介绍Flink 的3种常用的operator（keyby、reduce和Aggregations）及以具体可运行示例进行说明.
如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。
本文除了maven依赖外，没有其他依赖。

本专题分为五篇，即：
【flink番外篇】1、flink的23种常用算子介绍及详细示例（1）- map、flatmap和filter
【flink番外篇】1、flink的23种常用算子介绍及详细示例（2）- keyby、reduce和Aggregations
【flink番外篇】1、flink的23种常用算子介绍及详细示例（3）-window、distinct、join等
【flink番外篇】1、flink的23种常用算子介绍及详细示例（4）- union、window join、connect、outputtag、cache、iterator、project
【flink番外篇】1、flink的23种常用算子介绍及详细示例（完整版）

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/245352.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Windows系列：Windows Server 2016搭建AD域服务

Windows系列：Windows Server 2016搭建AD域服务

Windows Server 2016搭建AD域服务 AD域安装实验简介实验环境服务端配置客户端配置1. 加域方法一IP地址配置第二台客户端加域 2. 加域方法二Windows10客户端加完域是需要重启计算机的&#xff0c;等待客户端重启成功 AD域安装 实验简介 AD的全称是Active Directory。AD域是Win…
阅读更多...
不懂编程，如何获取全面海量的重要数据？

不懂编程，如何获取全面海量的重要数据？

在大数据和人工智能时代&#xff0c;数据的重要性变得更加突出。以下是数据在这个时代的重要性所体现的几个方面&#xff1a; 决策依据 模型训练 个性化服务 创新驱动 智能决策支持 本文&#xff0c;将介绍两个获取数据的方法 1、利用爬虫框架写采集程序 在前面&#xff…
阅读更多...
ProEasy机械手案例：电池边包胶

ProEasy机械手案例：电池边包胶

如下图所示&#xff0c;对一个电池三边包边&#xff0c;因客户现场有很多规格电池的大小&#xff0c;所以就需要建立动态的工具坐标来实现适配所有种类的电池 程序如下&#xff1a;Ddome程序 function Speed(num) --速度设置 MaxSpdL(2000) --movl最大速度…
阅读更多...
Apache Hive（部署+SQL）

Apache Hive（部署+SQL）

Hive架构 Hive部署 VMware虚拟机部署 一、在node1节点安装mysql数据库 二、配置Hadoop 三、下载 解压Hive 四、提供mysql Driver驱动 五、配置Hive 六、初始化元数据库 七、启动Hive(Hadoop用户) chown -R hadoop:hadoop apache-hive-3.1.3-bin hive 阿里云部…
阅读更多...
netcore swagger 错误 Failed to load API definition

netcore swagger 错误 Failed to load API definition

后端接口报错如下&#xff1a; 前端nswag报错如下&#xff1a; 根据网上查询到的资料说明&#xff0c;说一般swagger这种错误都是控制器里有接口代码异常造成的&#xff0c;通常是接口没有加属性Attribute&#xff0c; 比如[HttpPost("Delete")]、[HttpGet("Del…
阅读更多...
【Echarts】基本的关系图

【Echarts】基本的关系图

基本的关系图 let data {nodes: [{name: 操作系统集团11223,category: 4 // index},{name: 操作系统集团123,category: 3 // index},{name: 操作系统集团center,category: 2 // index},{name: 操作系统集团,category: 0 // index},{name: 浏览器有限公司,category: 0 // inde…
阅读更多...
双指针算法： 快乐数 与 盛水最多的容器

双指针算法： 快乐数 与 盛水最多的容器

&#x1f388;个人主页:&#x1f388; :✨✨✨初阶牛✨✨✨ &#x1f43b;推荐专栏1: &#x1f354;&#x1f35f;&#x1f32f;C语言初阶 &#x1f43b;推荐专栏2: &#x1f354;&#x1f35f;&#x1f32f;C语言进阶 &#x1f511;个人信条: &#x1f335;知行合一 前言 声明…
阅读更多...
医院不良事件报告系统源码带鱼骨图分析

医院不良事件报告系统源码带鱼骨图分析

医院不良事件上报系统通过 “事前的人员知识培训管理和制度落地促进”、“事中的事件上报和跟进处理”、 以及 “事后的原因分析和工作持续优化”&#xff0c;结合预存上百套已正在使用的模板&#xff0c;帮助医院从对护理事件、药品事件、医疗器械事件、医院感染事件、输血事件…
阅读更多...
32.768KHz时钟RTC晶振精度PPM值及频差计算

32.768KHz时钟RTC晶振精度PPM值及频差计算

一个数字电路就像一所城市的交通&#xff0c;晶振的作用就是十字路口的信号灯&#xff0c;因此晶振的品质及其电路应用尤其关键。数字电路又像生命体&#xff0c;它的运行就像人身体里的血液流通&#xff0c;它不是由单一的某个器件或器件单元构成&#xff0c;而是由多个器件及…
阅读更多...
力扣7.整数反转

力扣7.整数反转

题目描述 代码 自己写的像屎山&#xff0c;虽然能通过&#xff0c;但多了很多不必要的代码。 class Solution {public int reverse(int x) {int count 0;int res 0;//用temp2记录x的正负int temp2 x;if(x < 0){x -x;}int temp x;while(temp ! 0){temp temp / 10;cou…
阅读更多...
10分钟带你学会python模块和包的使用

10分钟带你学会python模块和包的使用

如果你用过 Python&#xff0c;那么你一定用过 import 关键字加载过各式各样的模块。但你是否熟悉 Python 中的模块与包的概念呢&#xff1f;或者&#xff0c;以下几个问题&#xff0c;你是否有明确的答案&#xff1f; 什么是模块&#xff1f;什么又是包&#xff1f;from matp…
阅读更多...
synchronized底层原理（一）

synchronized底层原理（一）

文章目录 1. 问题引入2. 相关概念3. Synchronized使用4. Synchronized底层原理1. 简介2. Monitor&#xff08;管程/监视器&#xff09;3. Java语言的内置管程synchronized4. Java对象的内存布局5. 如何使用MarkWord记录锁状态6. 偏向锁7. 轻量级锁 1. 问题引入 假设我们有1000…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023