Flink中Window详解之Window的聚合函数AggregateFunction

和 ReduceFunction 相似，AggregateFunction 也是基于中间状态计算结果的增量计算函数，但 AggregateFunction 在窗口计算上更加通用。AggregateFunction 接口相对 ReduceFunction 更加灵活，实现复杂度也相对较高。AggregateFunction 接口中定义了三个需要复写的方法，其中 add()定义数据的添加逻辑，getResult 定义了根据 accumulator 计算结果的逻辑，merge 方法定义合并 accumulator 的逻辑。

package windowimport org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector/*** @Author yqq* @Date 2021/12/27 20:42* @Version 1.0*/
case class StationLog(sid:String,callOut:String,callInput:String,callType:String,callTime:Long,duration:Long)
object AggregatFunctionTest {def main(args: Array[String]): Unit = {val environment = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.streaming.api.scala._//每隔5秒统计最近8秒内，每个基站的日志数量//读取数据源val stream: DataStream[StationLog] = environment.socketTextStream("node1", 8888).map(line => {val arr: Array[String] = line.split(",")new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)})//开窗stream.map(log=>{(log.sid,1)}).keyBy(_._1).window(SlidingProcessingTimeWindows.of(Time.seconds(8),Time.seconds(5)))//开窗，滑动窗口.aggregate(new MyAggregateFuntion,new MyWindowFunction).print()environment.execute()}//MyWindowFunction 输入数据来自于 MyAggregateFuntion，在窗口结束的时候先执行MyAggregateFuntion对象的getResult，然后在执行apply方法class MyWindowFunction extends WindowFunction[Long,(String,Long),String,TimeWindow] {override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[(String, Long)]): Unit = {out.collect((key,input.iterator.next()))//next得到的第一个值，迭代器中只有一个值}}class MyAggregateFuntion extends AggregateFunction[(String,Int),Long,Long] {//初始化一个累加器，开始的时候为0override def createAccumulator(): Long = 0//来一条数据执行一次override def add(value: (String, Int), accumulator: Long): Long = accumulator+value._2//在窗口结束的时候执行一次override def getResult(accumulator: Long): Long = accumulatoroverride def merge(a: Long, b: Long): Long = a+b}
}

Flink中Window详解之Window的聚合函数AggregateFunction相关推荐

Python中lambda详解（包括内置函数map、reduce、filter、sorted、max）
文章目录一.lambda是什么? 1.lambda语法 2.语法详解二.lambda的使用 1.定义 2.调用 3.替换 4.作返回值三.lambda作参数 1.map函数 2.reduce函数 ...
【转】图形流水线中坐标变换详解：模型矩阵、视角矩阵、投影矩阵
转自:图形流水线中坐标变换详解:模型矩阵.视角矩阵.投影矩阵_sherlockreal的博客-CSDN博客_视角矩阵图形流水线中坐标变换详解:模型矩阵.视角矩阵.投影矩阵图形流水线中坐标变换过程 ...
android中getSystemService详解
原文地址:android中getSystemService详解作者:邹斌 http://blog.sina.com.cn/s/blog_71d1e4fc0100o8qr.html http://blo ...
19. linux中权限详解，Linux权限位，读写执行权限真正含义，chmod详解
linux中权限详解,Linux权限位,读写执行权限真正含义,chmod详解文章目录 Linux权限位读写执行三种权限真正含义和作用权限对文件的作用权限对目录的作用示例 chmod 使用数 ...
浏览器中location详解
浏览器中location详解 window.location对象用于获取当前页面的URL信息. 属性解析 1. href 当前页面的URL. 比如访问github.com,在控制台中输入locatio ...
图形流水线中坐标变换详解：模型矩阵、视角矩阵、投影矩阵
图形流水线中坐标变换详解:模型矩阵.视角矩阵.投影矩阵图形流水线中坐标变换过程模型矩阵:模型局部坐标系和世界坐标系之间的桥梁 1.模型局部坐标系存在的意义 2.根据模型局部坐标系中点求其在世界坐标 ...
函数中{}输出格式详解(C#)
Console.WriteLine()函数中{}输出格式详解(C#) Console.WriteLine()函数的格式一直没怎么注意.今天同事问起Console.WriteLine({0:D3},a) ...
Java中CAS详解
转载自 Java中CAS详解在JDK 5之前Java语言是靠synchronized关键字保证同步的,这会导致有锁锁机制存在以下问题: (1)在多线程竞争下,加锁.释放锁会导致比较多的上下文切换 ...
C++学习笔记章节中面向对象详解
C++ 类&对象 C++类定义本质上是一个数据类型的蓝图,定义了类的对象包含的信息,以及可以在这个类对象上执行哪些操作. 类的定义是以class开头,后面接类的名称. 类的主体是包含在一个花 ...

Flink中Window详解之Window的聚合函数AggregateFunction

Flink中Window详解之Window的聚合函数AggregateFunction相关推荐

最新文章

热门文章