Flink Window中典型的增量聚合函数(ReduceFunction / AggregateFunction)

2023-12-14 22:52
文章标签 函数 window flink 增量 聚合 典型 reducefunction aggregatefunction

本文主要是介绍Flink Window中典型的增量聚合函数(ReduceFunction / AggregateFunction)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、什么是增量聚合函数

在Flink Window中定义了窗口分配器，我们只是知道了数据属于哪个窗口，可以将数据收集起来了；至于收集起来到底要做什么，其实还完全没有头绪，这也就是窗口函数所需要做的事情。所以在窗口分配器之后，我们还要再接上一个定义窗口如何进行计算的操作，这就是所谓的“窗口函数”（window functions）。
窗口可以将数据收集起来，最基本的处理操作当然就是基于窗口内的数据进行聚合。
我们可以每来一个数据就在之前结果上聚合一次，这就是“增量聚合”。
典型的增量聚合函数有两个：ReduceFunction 和 AggregateFunction。
在这里插入图片描述

二、归约函数ReduceFunction

源码解析

@FunctionalInterface
@Public
public interface ReduceFunction<T> extends Function, Serializable {T reduce(T var1, T var2) throws Exception;
}

实际案例
在Flink中，使用socket模拟实时的数据流DataStream，通过定义一个滚动窗口，窗口的大小为10s，按照id分区，使用reduce聚合函数实现value的累加统计

package com.flink.DataStream.WindowFunctions;import com.flink.POJOs.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
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.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;public class FlinkWindowReduceFunction {public static void main(String[] args) throws Exception {StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();streamExecutionEnvironment.setParallelism(1);DataStreamSource<String> streamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);// 注意这里为什么返回的是KeyedStream(建控流/分区流)，而不是DataStreamKeyedStream<WaterSensor, String> keyedStream = streamSource// 使用map函数将输入的string转为一个WaterSensor类.map(new MapFunction<String, WaterSensor>() {@Overridepublic WaterSensor map(String s) throws Exception {// 这里写的很详细，如何把string转为的WaterSensor类String[] strings = s.split(",");String id = strings[0];Long ts = Long.valueOf(strings[1]);Integer vc = Integer.valueOf(strings[2]);WaterSensor waterSensor = new WaterSensor();waterSensor.setId(id);waterSensor.setTs(ts);waterSensor.setVc(vc);return waterSensor;//return new WaterSensor(strings[0],Long.valueOf(strings[1]),Integer.valueOf(strings[2])}})// 按照id做keyBy分区（提问：KeyBy是如何实现分区的？）.keyBy(new KeySelector<WaterSensor, String>() {// 也可以直接使用lamda表达式更简单@Overridepublic String getKey(WaterSensor waterSensor) throws Exception {// getId()方法就是return的waterSensor.idreturn waterSensor.getId();}});/*** 窗口操作主要有两个部分：窗口分配器（Window Assigners）和窗口函数（WindowFunctions）* .window()方法需要传入一个窗口分配器,它指明了窗口的类型* */SingleOutputStreamOperator<WaterSensor> outputStreamOperator = keyedStream// 设置滚动窗口的大小（10秒）.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))// 使用匿名函数实现增量聚合函数ReduceFunction.reduce(new ReduceFunction<WaterSensor>() {@Overridepublic WaterSensor reduce(WaterSensor waterSensor1, WaterSensor waterSensor2) throws Exception {System.out.println("调用reduce方法，之前的结果:" + waterSensor1 + ",现在来的数据:" + waterSensor2);return new WaterSensor(waterSensor1.getId(), System.currentTimeMillis(), waterSensor1.getVc() + waterSensor2.getVc());}});outputStreamOperator.print();streamExecutionEnvironment.execute();}
}

启动Flink程序，启动nc，模拟输入

nc -lk 8888
# 00-10秒输入
a,11111,1
# 11-20秒输入
a,11111,2
a,22222,3
# 21-30秒输入
a,11111,4

查看控制台打印结果

WaterSensor{id='a', ts=11111, vc=1}
调用reduce方法，之前的结果:WaterSensor{id='a', ts=11111, vc=2},现在来的数据:WaterSensor{id='a', ts=22222, vc=3}
WaterSensor{id='a', ts=1702022598011, vc=5}
WaterSensor{id='a', ts=11111, vc=4}

在这里插入图片描述

三、聚合函数AggregateFunction

虽然ReduceFunction 可以解决大多数归约聚合的问题，但是我们通过上述案例可以发现：这个接口有一个限制，就是聚合状态的类型、输出结果的类型都必须和输入数据类型一样。
Flink Window API 中的 aggregate 就突破了这个限制，可以定义更加灵活的窗口聚合操作。这个方法需要传入一个 AggregateFunction 的实现类作为参数。AggregateFunction 可以看作是 ReduceFunction 的通用版本，这里有三种类型：输入类型（IN）、累加器类型（ACC）和输出类型（OUT）。输入类型 IN 就是输入流中元素的数据类型；累加器类型 ACC 则是我们进行聚合的中间状态类型；而输出类型当然就是最终计算结果的类型了。

源码解析

@PublicEvolving
public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {ACC createAccumulator();ACC add(IN var1, ACC var2);OUT getResult(ACC var1);ACC merge(ACC var1, ACC var2);
}

接口中有四个方法：
1.createAccumulator()
创建一个累加器，这就是为聚合创建了一个初始状态，每个聚合任务只会调用一次。
2.add()
将输入的元素添加到累加器中。
3.getResult()
从累加器中提取聚合的输出结果。
4.merge()
合并两个累加器，并将合并后的状态作为一个累加器返回。
所以可以看到，AggregateFunction 的工作原理是：首先调用 createAccumulator()为任务初始化一个状态（累加器）；而后每来一个数据就调用一次 add()方法，对数据进行聚合，得到的结果保存在状态中；等到了窗口需要输出时，再调用 getResult()方法得到计算结果。很明显，与 ReduceFunction 相同，AggregateFunction 也是增量式的聚合；而由于输入、中间状态、输出的类型可以不同，使得应用更加灵活方便。

案例解析

package com.flink.DataStream.WindowFunctions;import com.flink.POJOs.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
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.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;public class FlinkWindowAggregateFunction {public static void main(String[] args) throws Exception {StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();streamExecutionEnvironment.setParallelism(1);DataStreamSource<String> streamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);KeyedStream<WaterSensor, String> keyedStream = streamSource.map(new MapFunction<String, WaterSensor>() {@Overridepublic WaterSensor map(String s) throws Exception {String[] split = s.split(",");return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));}}).keyBy((KeySelector<WaterSensor, String>) waterSensor -> waterSensor.getId());// 窗口分配器WindowedStream<WaterSensor, String, TimeWindow> windowAssigner = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(20)));SingleOutputStreamOperator<String> aggregate = windowAssigner.aggregate(new AggregateFunction<WaterSensor, Integer, String>() {@Overridepublic Integer createAccumulator() {System.out.println("创建累加器");return 0;}@Overridepublic Integer add(WaterSensor value, Integer accumulator) {System.out.println("调用add方法,value=" + value);return accumulator + value.getVc();}@Overridepublic String getResult(Integer accumulator) {System.out.println("调用getResult方法");return accumulator.toString();}@Overridepublic Integer merge(Integer integer, Integer acc1) {System.out.println("调用merge方法");return null;}});aggregate.print();streamExecutionEnvironment.execute();}}

启动Flink程序，启动nc，模拟数据流输入
在这里插入图片描述在这里插入图片描述

创建累加器
调用add方法,value=WaterSensor{id='a', ts=1111, vc=1}
调用getResult方法
1
创建累加器
调用add方法,value=WaterSensor{id='a', ts=1111, vc=2}
调用add方法,value=WaterSensor{id='a', ts=1111, vc=3}
调用add方法,value=WaterSensor{id='a', ts=1111, vc=4}
调用getResult方法
9
创建累加器
调用add方法,value=WaterSensor{id='a', ts=1111, vc=5}
调用getResult方法
5

这篇关于Flink Window中典型的增量聚合函数(ReduceFunction / AggregateFunction)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/494172。

相关文章

poj2505（典型博弈）

poj2505（典型博弈）

题意：n = 1，输入一个k，每一次n可以乘以[2,9]中的任何一个数字，两个玩家轮流操作，谁先使得n >= k就胜出 这道题目感觉还不错，自己做了好久都没做出来，然后看了解题才理解的。 解题思路：能进入必败态的状态时必胜态，只能到达胜态的状态为必败态，当n >= K是必败态，[ceil(k/9.0),k-1]是必胜态， [ceil(ceil(k/9.0)/2.0),ceil(k/9.
阅读更多...
hdu1171（母函数或多重背包）

hdu1171（母函数或多重背包）

题意：把物品分成两份，使得价值最接近 可以用背包，或者是母函数来解，母函数（1 + x^v+x^2v+.....+x^num*v）（1 + x^v+x^2v+.....+x^num*v）（1 + x^v+x^2v+.....+x^num*v） 其中指数为价值，每一项的数目为（该物品数+1）个 代码如下： #include<iostream>#include<algorithm>
阅读更多...
C++操作符重载实例(独立函数)

C++操作符重载实例(独立函数)

C++操作符重载实例，我们把坐标值CVector的加法进行重载，计算c3=c1+c2时，也就是计算x3=x1+x2，y3=y1+y2，今天我们以独立函数的方式重载操作符+（加号），以下是C++代码： c1802.cpp源代码： D:\YcjWork\CppTour>vim c1802.cpp #include <iostream>using namespace std;/*** 以独立函数
阅读更多...
函数式编程思想

函数式编程思想

我们经常会用到各种各样的编程思想，例如面向过程、面向对象。不过笔者在该博客简单介绍一下函数式编程思想. 如果对函数式编程思想进行概括，就是f(x) = na(x) , y=uf(x)…至于其他的编程思想，可能是y=a(x)+b(x)+c(x)…，也有可能是y=f(x)=f(x)/a + f(x)/b+f(x)/c… 面向过程的指令式编程 面向过程，简单理解就是y=a(x)+b(x)+c(x)
阅读更多...
Jenkins构建Maven聚合工程，指定构建子模块

Jenkins构建Maven聚合工程，指定构建子模块

一、设置单独编译构建子模块 配置： 1、Root POM指向父pom.xml 2、Goals and options指定构建模块的参数： mvn -pl project1/project1-son -am clean package 单独构建project1-son项目以及它所依赖的其它项目。 说明： mvn clean package -pl 父级模块名/子模块名 -am参数
阅读更多...
利用matlab bar函数绘制较为复杂的柱状图，并在图中进行适当标注

利用matlab bar函数绘制较为复杂的柱状图，并在图中进行适当标注

示例代码和结果如下：小疑问：如何自动选择合适的坐标位置对柱状图的数值大小进行标注？😂 clear; close all;x = 1:3;aa=[28.6321521955954 26.2453660695847 21.69102348512086.93747104431360 6.25442246899816 3.342835958564245.51365061796319 4.87
阅读更多...
OpenCV结构分析与形状描述符（11）椭圆拟合函数fitEllipse()的使用

OpenCV结构分析与形状描述符（11）椭圆拟合函数fitEllipse()的使用

操作系统：ubuntu22.04 OpenCV版本：OpenCV4.9 IDE:Visual Studio Code 编程语言：C++11 算法描述 围绕一组2D点拟合一个椭圆。 该函数计算出一个椭圆，该椭圆在最小二乘意义上最好地拟合一组2D点。它返回一个内切椭圆的旋转矩形。使用了由[90]描述的第一个算法。开发者应该注意，由于数据点靠近包含的 Mat 元素的边界，返回的椭圆/旋转矩形数据
阅读更多...
Unity3D 运动之Move函数和translate

Unity3D 运动之Move函数和translate

CharacterController.Move 移动 function Move (motion : Vector3) : CollisionFlags Description描述 A more complex move function taking absolute movement deltas. 一个更加复杂的运动函数，每次都绝对运动。 Attempts to
阅读更多...
✨机器学习笔记（二）—— 线性回归、代价函数、梯度下降

✨机器学习笔记（二）—— 线性回归、代价函数、梯度下降

1️⃣线性回归（linear regression） f w , b ( x ) = w x + b f_{w,b}(x) = wx + b fw,b​(x)=wx+b 🎈A linear regression model predicting house prices： 如图是机器学习通过监督学习运用线性回归模型来预测房价的例子，当房屋大小为1250 f e e t 2 feet^
阅读更多...
JavaSE（十三）——函数式编程（Lambda表达式、方法引用、Stream流）

JavaSE（十三）——函数式编程（Lambda表达式、方法引用、Stream流）

函数式编程 函数式编程 是 Java 8 引入的一个重要特性，它允许开发者以函数作为一等公民（first-class citizens）的方式编程，即函数可以作为参数传递给其他函数，也可以作为返回值。 这极大地提高了代码的可读性、可维护性和复用性。函数式编程的核心概念包括高阶函数、Lambda 表达式、函数式接口、流（Streams）和 Optional 类等。 函数式编程的核心是Lambda
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2