一篇文章搞懂Flink的Window

本文隶属于专栏《1000个问题搞定大数据技术体系》&＃xff0c;该专栏为笔者原创&＃xff0c;引用请注明来源&＃xff0c;不足和错误之处请在评论区帮忙指出&＃xff0c;谢谢&＃xff01;

本专栏目录结构和参考文献请见1000个问题搞定大数据技术体系

为什么需要 window &＃xff1f;

对于流式处理&＃xff0c;如果我们需要求取总和&＃xff0c;平均值&＃xff0c;或者最大值&＃xff0c;最小值等&＃xff0c;是做不到的&＃xff0c;因为数据一直在源源不断的产生&＃xff0c;即数据是没有边界的&＃xff0c;所以没法求最大值&＃xff0c;最小值&＃xff0c;平均值等&＃xff0c;所以为了一些数值统计的功能&＃xff0c;我们必须指定时间段&＃xff0c;对某一段时间的数据求取一些数据值是可以做到的。或者对某一些数据求取数据值也是可以做到的&＃xff0c;所以&＃xff0c;流上的聚合需要由 window 来划定范围&＃xff0c;比如 “计算过去的5分钟” &＃xff0c;或者 “最后100个元素的和” 。

什么是 window &＃xff1f;

window是一种可以把无限数据切割为有限数据块的手段

窗口可以是 时间驱动的 【Time Window】&＃xff08;比如&＃xff1a;每30秒&＃xff09;或者 数据驱动的【Count Window】 &＃xff08;比如&＃xff1a;每100个元素&＃xff09;。

1、窗口的基本类型介绍

窗口通常被区分为不同的类型:
tumbling windows&＃xff1a;滚动窗口 【没有重叠】
sliding windows&＃xff1a;滑动窗口 【有重叠】
session windows&＃xff1a;会话窗口 &＃xff0c;一般没人用

tumbling windows类型&＃xff1a;没有重叠的窗口

sliding windows&＃xff1a;滑动窗口 【有重叠】

2、Flink的窗口介绍

Time Window窗口的应用

time window又分为滚动窗口和滑动窗口&＃xff0c;这两种窗口调用方法都是一样的&＃xff0c;都是调用timeWindow这个方法&＃xff0c;如果传入一个参数就是滚动窗口&＃xff0c;如果传入两个参数就是滑动窗口

Count Windos窗口的应用

与timeWindow类型&＃xff0c;CountWinodw也可以分为滚动窗口和滑动窗口&＃xff0c;这两个窗口调用方法一样&＃xff0c;都是调用countWindow&＃xff0c;如果传入一个参数就是滚动窗口&＃xff0c;如果传入两个参数就是滑动窗口

自定义window的应用

如果time window和 countWindow还不够用的话&＃xff0c;我们还可以使用自定义window来实现数据的统计等功能。

3、window的数值聚合统计

对于某一个window内的数值统计&＃xff0c;我们可以增量的聚合统计或者全量的聚合统计

增量聚合统计&＃xff1a;

窗口当中每加入一条数据&＃xff0c;就进行一次统计
•reduce(reduceFunction)
•aggregate(aggregateFunction)
•sum(),min(),max()

需求&＃xff1a;通过接收socket当中输入的数据&＃xff0c;统计每5秒钟数据的累计的值

代码实现&＃xff1a;

package com.shockang.study.bigdata.flink.windowimport org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.datastream.DataStreamSink
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Timeobject FlinkTimeCount {def main(args: Array[String]): Unit &＃61; {val environment: StreamExecutionEnvironment &＃61; StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._val socketStream: DataStream[String] &＃61; environment.socketTextStream("node01", 9000)val print: DataStreamSink[(Int, Int)] &＃61; socketStream.map(x &＃61;> (1, x.toInt)).keyBy(0).timeWindow(Time.seconds(5)).reduce(new ReduceFunction[(Int, Int)] {override def reduce(t: (Int, Int), t1: (Int, Int)): (Int, Int) &＃61; {(t._1, t._2 &＃43; t1._2)}}).print()environment.execute("startRunning")}
}


全量聚合统计&＃xff1a;

等到窗口截止&＃xff0c;或者窗口内的数据全部到齐&＃xff0c;然后再进行统计&＃xff0c;可以用于求窗口内的数据的最大值&＃xff0c;或者最小值&＃xff0c;平均值等
等属于窗口的数据到齐&＃xff0c;才开始进行聚合计算【可以实现对窗口内的数据进行排序等需求】
apply(windowFunction)
process(processWindowFunction)

processWindowFunction比windowFunction提供了更多的上下文信息。

需求&＃xff1a;通过全量聚合统计&＃xff0c;求取每3条数据的平均值

package com.shockang.study.bigdata.flink.windowimport org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.datastream.DataStreamSink
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.util.Collectorobject FlinkCountWindowAvg {def main(args: Array[String]): Unit &＃61; {val environment: StreamExecutionEnvironment &＃61; StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._val socketStream: DataStream[String] &＃61; environment.socketTextStream("node01", 9000)//统计一个窗口内的数据的平均值val socketDatas: DataStreamSink[Double] &＃61; socketStream.map(x &＃61;> (1, x.toInt)).keyBy(0)//.timeWindow(Time.seconds(10)).countWindow(3)//通过process方法来统计窗口的平均值.process(new MyProcessWindowFunctionclass).print()//必须调用execute方法&＃xff0c;否则程序不会执行environment.execute("count avg")}
}/** ProcessWindowFunction 需要跟四个参数* 输入参数类型&＃xff0c;输出参数类型&＃xff0c;聚合的key的类型&＃xff0c;window的下界**/
class MyProcessWindowFunctionclass extends ProcessWindowFunction[(Int, Int), Double, Tuple, GlobalWindow] {override def process(key: Tuple, context: Context, elements: Iterable[(Int, Int)], out: Collector[Double]): Unit &＃61; {var totalNum &＃61; 0;var countNum &＃61; 0;for (data <- elements) {totalNum &＃43;&＃61; 1countNum &＃43;&＃61; data._2}out.collect(countNum / totalNum)}
}