提醒：本文的示例代码基于flink1.13，在讲window的使用时也会说明flink版本一些api的弃用情况。
一、Time的简介
flink 中的streaming定义了多种流式处理的时间，Event Time（事件时间）、Ingestion Time（接收时间）、Processing Time（处理时间）。
Event Time：指事件产生的时间，比如业务数据库一条数据产生的时间、一条日志数据产生的时间等。
Ingestion Time：指flink接收数据的时间。
Processing Time：指数据被flink算子处理的时间。
在真实的业务代码开发中，我们常使用Event TIme、Processing Time。
二、Window的概念
flink中，streaming流式计算被设计为用于处理无限数据集的数据处理引擎，其中无限数据集是指一种源源不断有数据过来的数据集，window （窗口）将无界数据流切割成为有界数据流进行处理的方式。实现方式是将流分发到有限大小的桶（bucket）中进行分析。
三、Window的类型
1、分类关系
* TimeWindow（计时窗口），按照一定时间生成 Window（比如：每5秒
* CountWindow（计数窗口），按照指定的数据量生成一个 Window，与时间无关（比如：每20个元素）。
* TumblingWindow（滚动窗口）
* Sliding Window（滑动窗口）
* Session Window（会话窗口）
它们之间的关系图
2、滚动窗口（Tumbling Windows）
将数据依据固定的窗口长度对数据进行切片，它的特点是时间对齐，窗口长度固定，没有重叠。例如：如果你指定了一个 5分钟大小的滚动窗口，窗口的创建如下图所示：
适合做每个时间段的聚合计算，例如：统计每5分钟内用户的热搜词
3、滑动窗口（Sliding Windows）
滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成，特点为时间对齐，窗口长度固定，可以有重叠。例如：你有 10 分钟的窗口和 5 分钟的滑动，那么每个窗口中 5 分钟的窗口里包含着上个 10 分钟产生的数据，如下图所示：
适用于对最近一个时间段内的统计，例如：求某接口最近 5min 的失败率来决定是否要报警这种场景。
4、会话窗口（Session Windows）
由一系列事件组合一个指定时间长度的 timeout 间隙组成，类似于 web 应用的session，也就是一段时间没有接收到新数据就会生成新的窗口。他的特点为：窗口大小是由数据本身决定，它没有固定的开始和结束时间。会话窗口根据Session gap间隙切分不同的窗口，当一个窗口在大于Session gap间隙的时间内没有接收到新数据时，窗口将关闭。例如：设置的时间gap是5秒，那么，当相邻的记录相差>=5秒时，则触发窗口。
适用于每个用户在一个独立的session中平均页面访问时长，前后两个session的间隔时间为15分钟这种场景。
四、windows 的使用
window算子api的使用分有key的、无key的，它们api分别的写法如下：
Keyed Windows
Non-Keyed Windows
我们下面按照上面的window分类关系去讲解api的使用，且举得例子都是Keyed Windows的，可以类比使用对应api理解Non-Keyed Windows。
1、Time Window
在time Window中，我们经常会在flink的老版本中使用timeWindow，如下图：
输入一个Time.seconds(n)是滚动窗口，输入两个是滑动窗口。特别注意，这里默认使用的Time是Processing Time。
在flink1.13中方法已经过时，能用但不建议使用。请使用原生window进行使用，如下图：
在window的参数中指定使用的窗口类型、时间类型，这个示例可以看出使用的是滚动窗口，时间类型为Processing Time
2、Count Window
Count Window中我们常在flink中使用
countWindow（高版本中没有过时），传入一个参数就是滚动窗口，如果传入两个参数就是滑动窗口，如下图：
3、自定义Window
如下图Keyed Windows、Non-Keyed Windows两种使用自定义的接口， 可以自己定义trigger（触发器）、evictor(移除器)、allowedLateness(允许窗口延时)、sideOutputLateData(侧输出流)等，自定义window。
4、示例
4.1 滚动窗口示例
需求：每隔5s时间，统计最近5s出现的单词
代码：
```
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
/**
* todo: 基于计时的滚动窗口应用--Tumbling Windows
*
* 滚动窗口，每隔5s时间，统计最近5s出现的单词
*/
object TestTimeWindowByTumbling {
def main(args: Array[String]): Unit = {
//todo:1、获取流式处理的环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//todo:2、获取数据源
val socketTextStream: DataStream[String] = env.socketTextStream("companynode01",19999)
//todo: 3、对数据进行操作处理
socketTextStream.flatMap(x=>x.split(" "))
.map(x=>(x, 1))
.keyBy(_._1)
// .timeWindow(Time.seconds(5)) //过时
// 滚动窗口
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.sum(1)
.print()
//todo: 4、启动
env.execute("TestTimeWindowByTumbling")
}
}
```
4.2 滑动窗口示例
需求：每隔5s时间，统计最近10s出现的单词
代码：
```
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
/**
* todo: 基于计时的滚动窗口应用--Sliding Windows
*
* 滑动窗口，每隔5s时间，统计最近10s出现的单词
*/
object TestTimeWindowBySliding {
def main(args: Array[String]): Unit = {
//todo:1、获取流式处理的环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//todo:2、获取数据源
val socketTextStream: DataStream[String] = env.socketTextStream("companynode01",19999)
//todo: 3、对数据进行操作处理
socketTextStream.flatMap(x=>x.split(" "))
.map(x=>(x, 1))
.keyBy(_._1)
// .timeWindow(Time.seconds(15),Time.seconds(5)) //过时
//滑动窗口
.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
.sum(1)
.print()
//todo: 4、启动
env.execute("TestTimeWindowBySliding")
}
}
```
五、window Function（窗口函数）
1、分类
窗口函数定义了要对窗口中收集的数据做的计算操作。主要可以分为两类：
* 增量聚合函数（incremental aggregation functions），它在每条数据到来就进行计算，保持一个简单的状态。典型的增量聚合函数有ReduceFunction、AggregateFunction等。
* 全量窗口函数（full window functions），它先把窗口所有数据收集起来，等到计算的时候会遍历所有数据。典型的全量窗口聚合函数有apply、process。
2、增量聚合统计
窗口当中每加入一条数据，就进行一次统计。常用的增量聚合算子有reduce(reduceFunction)、aggregate(aggregateFunction)sum()、min()、max()等。
示例：
需求：通过接收socket当中输入的单词，统计每5秒钟单词的累计数量
使用reduce的代码
```
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
//import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
/**
* 增量聚合函数
* 通过接收socket当中输入的数据，统计每5秒钟数据的累计的值
* 基于 `reduce` 函数的计时窗口数据增量聚合
*/
object TestReduceOfTimeWindow {
def main(args: Array[String]): Unit = {
val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val socketTextStream: DataStream[String] = env.socketTextStream("companynode01", 19999)
socketTextStream.flatMap(x => x.split(" "))
.map(x=>(x, 1))
// .keyBy(0)
.keyBy(_._1)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.reduce((c1, c2) => (c1._1, c1._2+c2._2))
.print()
env.execute()
}
}
```
使用aggregate的代码
```
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.common.functions.AggregateFunction
/**
* 增量聚合函数
* 基于`aggregate`函数的计时窗口数据增量聚合
*/
object TestAggregateOfTimeWindow {
def main(args: Array[String]): Unit = {
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val socketTextStream: DataStream[String] = env.socketTextStream("companynode01", 9999)
socketTextStream.flatMap(x => x.split(" "))
.map(x => (x, 1))
.keyBy(_._1)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.aggregate(new MyAggregateFunction)
.print()
env.execute("TestAggregateOfTimeWindow")
}
}
//自定义AggregateFunction函数
class MyAggregateFunction extends AggregateFunction[(String, Int), (String, Int), (String, Int)] {
var initAccumulator: (String, Int) = ("", 0)
//累加值的初始化操作
override def createAccumulator(): (String, Int) = {
initAccumulator
}
//累加元素
override def add(in: (String, Int), acc: (String, Int)): (String, Int) = {
(in._1, acc._2 + in._2)
}
// 聚合的结果
override def getResult(acc: (String, Int)): (String, Int) = {
acc
}
//分布式累加
override def merge(acc: (String, Int), acc1: (String, Int)): (String, Int) = {
(acc._1, acc._2 + acc1._2)
}
}
```
3、全量聚合统计
等到窗口截止，或者窗口内的数据全部到齐，然后再进行统计。
常用的增量聚合算子有apply(windowFunction)、process(processWindowFunction)其中processWindowFunction 比 windowFunction 提供了更多的上下文信息。
示例：
需求：通过接收socket当中输入的数值，统计5秒钟输入数值的平均值。
使用apply函数实现的代码：
```
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
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
import org.apache.flink.util.Collector
object TestApplyOfTimeWindow {
def main(args: Array[String]): Unit = {
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//2、获取数据源，例如输入如下数据源
/**
* 1
* 2
* 3
* 4
* 5
* 6
*/
val socketStream: DataStream[String] = env.socketTextStream("companynode01", 19999)
socketStream.flatMap(x => x.split(" "))
.map(x => ("countAvg", x.toInt))
// .keyBy(0)
//keyBy中的key是个虚拟key，不会输出
.keyBy(_._1)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.apply(new MyApplyWindowFunction)
.print()
env.execute("TestApplyOfTimeWindow")
}
}
class MyApplyWindowFunction extends WindowFunction[(String, Int), Double, String, TimeWindow]{
override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[Double]): Unit = {
//记次数
var totalNum = 0
//记累加结果
var countNum = 0
for(elem <- input) {
totalNum += 1
countNum += elem._2
}
out.collect(countNum/totalNum)
}
}
```
使用process函数实现的代码：
```
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
/**
* 基于 `process` 函数的计时窗口数据全量聚合
*/
object TestProcessOfTimeWindow {
def main(args: Array[String]): Unit = {
val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val socketStream: DataStream[String] = env.socketTextStream("companynode01", 19999)
socketStream.flatMap(x => x.split(" "))
.map(x => ("countAvg", x.toInt))
.keyBy(x => x._1)
//是使用的processingTime,接口已经过时
// .timeWindow(Time.seconds(5))
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new MyProcessWindowFunction)
.print()
env.execute("TestProcessOfTimeWindow")
}
}
class MyProcessWindowFunction extends ProcessWindowFunction[(String, Int), Double, String, TimeWindow] {
override def process(key: String, context: Context,
elements: Iterable[(String, Int)],
out: Collector[Double]): Unit = {
//计次数
var totalNum = 0
//计累加次数
var countNum = 0
for (elem <- elements) {
totalNum += 1
countNum += elem._2
}
//计算平均值, totalNum.asInstanceOf[Double]是将totalNum强转成Double类型
out.collect(countNum/totalNum.asInstanceOf[Double])
}
}
```
