1、 概述

• 窗口 Window

  • 流数据计算中一般对数据尽心操作之前都会先进行开窗，即基于一个什么样的窗口上做这个计算
  • Flink 提供了开箱即用的各种窗口，比如滑动窗口、滚动窗口、会话窗口以及非常灵活的自定义窗口

• 时间 Time

  • Flink 中窗口计算，基本都是基于时间窗口设置
  • Flink 实现了 Watermark 的机制，能够支持基于事件时间的处理，能够容忍迟到、乱序的数据

• 状态 State

  • Flink计算引擎，自身就是基于状态计算框架，默认情况下程序自己管理状态
  • 提供一致性的语义，使得用户在编程时能够更轻松、更容易地去管理状态
  • 提供一套非常简单明了的 State API，包括ValueState、ListState、MapState，BroadcastState

• 检查点 Checkpoint

  • Flink Checkpoint 检查点：保存状态数据
  • 基于 Chandy-Lamport 算法实现了一个分布式的一致性的快照，从而提供了一致性的语义
  • 进行 Checkpoint 后，可以设置自动进行故障恢复
  • 保存点 Savepoint，人工进行 Checkpoint 操作，进行程序恢复执行

2、 Flink 的 Window 和 Time

2.1、Window API

在 Flink 流计算中，提供 Window 窗口 API 分为 2 种：

• 针对 KeyedStream 窗口 API
  Window 先对数据流 DataStream 进行分组 keyBy ，再设置窗口 Window，最后进行聚合 apply 操作。
  • 第一步、数据流 DataStream 调用 keyBy 函数分组，获取 KeyedStream
  • 第二步、KeyedStream.window 设置窗口
  • 第三步、聚合操作，对窗口中数据进行聚合统计，函数：reduce、aggregate、apply() 等。

stream.keyBy(...)          <-  keyed versus non-keyed windows.window(...)         <-  required: "assigner"[.trigger(...)]       <-  optional: "trigger" (else default trigger)[.evictor(...)]       <-  optional: "evictor" (else no evictor)[.allowedLateness()]  <-  optional, else zero.reduce/fold/apply() <-  required: "function"

• 针对 KeyedStream 窗口 API
  • 直接调用窗口函数：windowAll，然后再对窗口所有数据进行处理，未进行分组；
  • 聚合操作，对窗口中数据进行聚合统计，函数：reduce、aggregate、apply() 等。

stream.windowAll(...)      <-  required: "assigner"[.trigger(...)]       <-  optional: "trigger" (else default trigger)[.evictor(...)]       <-  optional: "evictor" (else no evictor)[.allowedLateness()]  <-  optional, else zero.reduce/fold/apply() <-  required: "function"

方括号 [ ] 内的命令是可选的，这表明 Flink 允许根据需求自定义 window 逻辑。使用 keyBy 的流,应该使用 window 方法，未使用 keyBy 的流,应该调用 windowAll 方法。

2.1.1、WindowAssigner

window/windowAll 方法接收的输入是一个 WindowAssigner， WindowAssigner 负责将每条输入的数据分发到正确的 window 中。如果需要自己定制数据分发策略，则可以实现一个 class，继承自 WindowAssigner。

2.1.2、Trigger

trigger 用来判断一个窗口是否需要被触发，每个 WindowAssigner 都自带一个默认的 trigger，如果默认的 trigger 不能满足你的需求，则可以自定义一个类，继承自Trigger 即可。

• onElement()
• onEventTime()
• onProcessingTime()

此抽象类的这三个方法会返回一个 TriggerResult， TriggerResult 有如下几种可能的选择：

• CONTINUE 不做任何事情
• FIRE 触发 window
• PURGE 清空整个 window 的元素并销毁窗口
• FIRE_AND_PURGE 触发窗口，然后销毁窗口

2.1.3、Evictor

evictor 主要用于做一些数据的自定义操作，可以在执行用户代码之前，也可以在执行用户代码之后。本接口提供了两个重要的方法，即 evicBefore 和 evicAfter两个方法。

Flink 提供了如下三种通用的 evictor：

• CountEvictor 保留指定数量的元素
• TimeEvictor 设定一个阈值 interval，删除所有不再 max_ts - interval 范围内的元素，其中 max_ts 是窗口内时间戳的最大值
• DeltaEvictor 通过执行用户给定的 DeltaFunction 以及预设的 theshold，判断是否删
  除一个元素。

2.2、窗口类型

Flink Window 窗口的结构中，有两个必须的两个操作：

• 第一、窗口分配器（WindowAssigner）：将数据流中的元素分配到对应的窗口。
• 第二、窗口函数（Window Function）：当满足窗口触发条件后，对窗口内的数据使用窗口处理函数（Window Function）进行处理，常用的有 reduce、aggregate、process。

在 Flink 窗口计算中，无论时间窗口还是计数窗口，都可以分为 2 种类型：滚动 Tumbling 和 滑动 Sliding 窗口。

• 滚动窗口（Tumbling Window）
  条件：窗口大小 size = 滑动间隔 slide

• 滚动窗口（Tumbling Window）
  条件：窗口大小 != 滑动间隔，通常条件【窗口大小 size > 滑动间隔 slide】

Window 的生命周期是什么？
简单的说，当有第一个属于该 window 元素到达时就创建了一个 window，当时间或事件触发该 windowremoved 的时候则结束。每个 window 都有一个 Trigger 和 一个 Function，function用于计算，tigger 用于触发 window 条件。同时也可以使用 Evictor 在 Trigger 触发前后对 window 的元素进行处理。

2.2.1、Tumbling Windows

滚动窗口分配器（Tumbling windows assigner）将每个元素分配给指定窗口大小的窗口。滚动窗口具有固定大小，不会重叠。例如，如果指定大小为 5 分钟的滚动窗口，则将评估当前窗口，并且每 5 分钟启动一个新窗口，如下图所示：

示例代码：

// 3-1. 对数据进行转换处理: 过滤脏数据，解析封装到二元组中SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = inputStream.filter(line -> line.trim().split(",").length == 2).map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String line) throws Exception {System.out.println("item: " + line);String[] array = line.trim().split(",");Tuple2<String, Integer> tuple = Tuple2.of(array[0], Integer.parseInt(array[1]));// 返回return tuple;}});// todo: 3-2. 窗口计算，每隔5秒计算最近5秒各个卡口流量SingleOutputStreamOperator<String> windowStream = mapStream// a. 设置分组key，按照卡口分组.keyBy(tuple -> tuple.f0)// b. 设置窗口，并且为滚动窗口：size=slide.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))// c. 窗口计算，窗口函数.apply(new WindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {// 定义变量，对日前时间数据进行转换private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss") ;@Overridepublic void apply(String key, TimeWindow window,Iterable<Tuple2<String, Integer>> input,Collector<String> out) throws Exception {// 获取窗口时间信息：开始时间和结束时间String winStart = this.format.format(window.getStart());String winEnd = this.format.format(window.getEnd()) ;// 对窗口中数据进行统计：求和int sum = 0 ;for (Tuple2<String, Integer> tuple : input) {sum += tuple.f1 ;}// 输出结果数据String output = "window: [" + winStart + " ~ " + winEnd + "], " + key + " = " + sum ;out.collect(output);}});

2.2.2、Sliding Windows

滑动窗口分配器（sliding windows assigner）将元素分配给固定长度的窗口。与滚动窗口分配器类似，窗口的大小由窗口大小参数配置。窗口滑动参数控制滑动窗口的启动频率。因此，如果 sliding小于size，则滑动窗口可能会重叠。在这种情况下，元素被分配给多个窗口。例如，可以有大小为 10 分钟的窗口，该窗口滑动 5 分钟。这样，您每 5 分钟就会得到一个窗口，其中包含过去 10 分钟内到达的事件，如下图所示：

示例代码：

// 3-1. 对数据进行转换处理: 过滤脏数据，解析封装到二元组中SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = inputStream.filter(line -> line.trim().split(",").length == 2).map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String line) throws Exception {System.out.println("item: " + line);String[] array = line.trim().split(",");return Tuple2.of(array[0], Integer.parseInt(array[1]));}});// todo: 3-2. 窗口计算，每隔5秒计算最近5秒各个卡口流量SingleOutputStreamOperator<String> windowStream = mapStream// a. 设置分组key，按照卡口分组.keyBy(tuple -> tuple.f0)// b. 设置窗口，并且为滚动窗口：size != slide.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))// c. 窗口计算，窗口函数.apply(new WindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {// 定义变量，对日前时间数据进行转换private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss") ;@Overridepublic void apply(String key, TimeWindow window,Iterable<Tuple2<String, Integer>> input,Collector<String> out) throws Exception {// 获取窗口时间信息：开始时间和结束时间String winStart = this.format.format(window.getStart());String winEnd = this.format.format(window.getEnd()) ;// 对窗口中数据进行统计：求和int sum = 0 ;for (Tuple2<String, Integer> tuple : input) {sum += tuple.f1 ;}// 输出结果数据String output = "window: [" + winStart + " ~ " + winEnd + "], " + key + " = " + sum ;out.collect(output);}});

2.2.3、Session Windows

会话窗口分配器（session windows assigner）按活动会话对元素进行分组。与滚动窗口和滑动窗口相比，会话窗口不重叠，也没有固定的开始和结束时间。相反，当会话窗口在一段时间内未收到元素时（即，当出现不活动间隙时），会话窗口将关闭。会话窗口分配器可以配置静态会话间隙或会话间隙提取器功能，该函数定义不活动时间的时间。当此时间段到期时，当前会话将关闭，后续元素将分配给新的会话窗口。

示例代码：

// 3-1. 过滤和转换数据类型SingleOutputStreamOperator<Integer> mapStream = inputStream.filter(line -> line.trim().length() > 0).map(new MapFunction<String, Integer>() {@Overridepublic Integer map(String value) throws Exception {System.out.println("item: " + value);return Integer.parseInt(value);}});// 3-2. 直接对DataStream流进行窗口操作SingleOutputStreamOperator<String> windowStream = mapStream// a. 设置窗口：会话窗口，超时时间为5秒.windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))// b. 设置窗口函数，对窗口中数据进行计算.apply(new AllWindowFunction<Integer, String, TimeWindow>() {// 定义变量，对日前时间数据进行转换private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss") ;@Overridepublic void apply(TimeWindow window, Iterable<Integer> values, Collector<String> out) throws Exception {// 获取窗口时间信息：开始时间和结束时间String winStart = this.format.format(window.getStart());String winEnd = this.format.format(window.getEnd()) ;// 对窗口中数据进行求和int sum = 0 ;for (Integer value : values) {sum += value ;}// 输出结果数据String output = "window: " + winStart + " ~ " + winEnd + " -> " + sum ;out.collect(output);}});

2.2.4、Global Windows

全局窗口分配器（global windows assigner）将具有相同键的所有元素分配给同一个全局窗口。只有自己自定义触发器的时候该窗口才能使用。否则，将不会执行任何计算，因为全局窗口没有一个自然的终点，我们可以在该端点处理聚合元素。

示例代码：

// 3-1. 过滤和转换数据类型SingleOutputStreamOperator<Integer> mapStream = inputStream.filter(line -> line.trim().length() > 0).map(new MapFunction<String, Integer>() {@Overridepublic Integer map(String value) throws Exception {System.out.println("item: " + value);return Integer.parseInt(value);}});// TODO： 3-2. 直接对DataStream流进行窗口操作SingleOutputStreamOperator<String> windowStream = mapStream// a. 设置窗口，滚动计数窗口.countWindowAll(5)// b. 设置窗口函数，计算窗口中数据.apply(new AllWindowFunction<Integer, String, GlobalWindow>() {@Overridepublic void apply(GlobalWindow window, Iterable<Integer> values, Collector<String> out) throws Exception {// 对窗口中数据进行求和int sum = 0 ;for (Integer value : values) {sum += value ;}// 输出累加求和值String output = "sum = " + sum ;out.collect(output);}});

2.3、Time 时间语义

• 事件时间 EventTime：事件真真正正发生产生的时间，比如订单数据中订单时间表示订单产生的时间；
• 摄入时间 IngestionTime：数据被流式程序获取的时间；
• 处理时间 ProcessingTime：事件真正被处理/计算的时间。

基于事件时间 EventTime 窗口分析，指定事件时间字段，使用 assignTimestampsAndWatermarks 方法，类型必须为 Long 类型。

// 3-1. 过滤脏数据和指定事件时间字段字段SingleOutputStreamOperator<String> timeStream = inputStream.filter(line -> line.trim().split(",").length == 3)// todo: step1、指定事件时间字段，并且数据类型为Long类型.assignTimestampsAndWatermarks(WatermarkStrategy// 暂不考虑数据乱序和延迟.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))// 指定事件时间字段.withTimestampAssigner(new SerializableTimestampAssigner<String>() {private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");@SneakyThrows@Overridepublic long extractTimestamp(String element, long recordTimestamp) {// 2022-04-01 09:00:01,a,1 -> 2022-04-01 09:00:01 -> 1648774801000System.out.println("element -> " + element);// 分割字符串String[] array = element.split(",");// 获取事件时间String eventTime = array[0];// 转换格式Date eventDate = format.parse(eventTime);// z转换Long类型并返回return eventDate.getTime();}}));

默认情况下（不考虑乱序和延迟），当数据事件时间EventTime >= 窗口结束时间，触发窗口数据计算。
基于事件时间EventTime窗口分析，如果不考虑数据延迟乱序，当窗口被触发计算以后，延迟乱序到达的数据将不会被计算，而是直接丢弃。

窗口起始时间计算方式：
timestamp - (timestamp - offset + wondowsize)%windowsize
如：00：00：01 窗口大小：5s 乱序时间：0s，则：
1 -（1 - 0 + 5 ）% 5 = 0

2.4、乱序和延迟数据处理

• Watermark 水印机制
  在实际业务数据中，数据乱序到达流处理程序，属于正常现象，原因在于网络延迟导致数据延迟，无法避免的，所以应该可以允许数据乱序达到（在某个时间范围内），依然参与窗口计算。
  

• Allowed Lateness 允许延迟
  默认情况下，当 watermark 超过 end-of-window 之后，再有之前的数据到达时，这些数据会被删除。为了避免有些迟到的数据被删除，因此产生了 allowedLateness 的概念。
  

• 乱序数据：Watermark，窗口数据计算等一下

  • 使用水位线Watermark，给每条数据加上一个时间戳
  • Watermark = 数据事件时间 - 最大允许乱序时间
  • 当数据的Watermark >= 窗口结束时间，并且窗口内有数据，触发窗口数据计算

• 延迟数据：AllowedLateness，窗口计算状态保存一段时间

  • 设置方法参数：allowedLateness，表示允许延迟数据最多可以迟到多久，还可以进行计算（保存窗口，并且触发窗口计算）
  • 当某个窗口触发计算以后，继续等待多长时间，如果在等待时间范围内，有数据达到时，依然会触发窗口计算。如果到达等待时长以后，没有数据达到，销毁窗口数据信息。

真正迟到的数据默认会被丢弃，可通过侧边流输出到文件：

• 1、窗口 window 的作用是为了周期性的获取数据；
• 2、watermark 作用是防止数据出现乱序(经常)，事件时间内获取不到指定的全部数据，做的一种保险方法；
• 3、allowLateNess 是将窗口关闭时间再延迟一段时间；
• 4、sideOutPut 是最后兜底操作，所有过期延迟数据，指定窗口已经彻底关闭，就会把数据放到侧输出流。

2.5、综合案例

public static void main(String[] args) throws Exception {// 1. 执行环境-envConfiguration configuration = new Configuration();configuration.setString("rest.port", "8081");StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);env.setParallelism(1) ;// todo: 设置CheckpointsetEnvCheckpoint(env) ;// todo: 设置重启策略env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));// 2. 数据源-sourceDataStreamSource<String> inputStream = env.socketTextStream("127.0.0.1", 9999);// 3. 数据转换-transformation/*业务数据：o_101,u_121,11.50,2022-04-05 10:00:023-1. 过滤、解析和封装数据3-2. 设置事假时间字段值和水位线Watermark3-3. 窗口设置及处理数据*/// 3-1. 过滤、解析和封装数据SingleOutputStreamOperator<OrderEvent> orderStream = inputStream.filter(line -> null != line && line.trim().split(",").length == 4).map(new MapFunction<String, OrderEvent>() {@Overridepublic OrderEvent map(String value) throws Exception {// 分割为单次String[] array = value.split(",");// 封装实体类对象OrderEvent orderEvent = new OrderEvent() ;orderEvent.setOrderId(array[0]);orderEvent.setUserId(array[1]);orderEvent.setOrderMoney(Double.parseDouble(array[2]));orderEvent.setOrderTime(array[3]);// 返回实例对象return orderEvent;}});// 3-2. 设置事假时间字段值和水位线WatermarkSingleOutputStreamOperator<OrderEvent> timeStream = orderStream.assignTimestampsAndWatermarks(WatermarkStrategy// 允许最大乱序时间：2秒，等待2秒钟触发窗口计算.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(2))// 获取订单时间，设置事件事假.withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");@SneakyThrows@Overridepublic long extractTimestamp(OrderEvent element, long recordTimestamp) {System.out.println("order -> " + element);// 获取订单时间String orderTime = element.getOrderTime();// 转换为Date日期类型Date orderDate = format.parse(orderTime);// 转换Long并返回return orderDate.getTime();}}));// 3-3. 窗口设置及处理数据OutputTag<OrderEvent> lateOutputTag = new OutputTag<OrderEvent>("late-order"){} ;SingleOutputStreamOperator<OrderReport> windowStream = timeStream// 按照用户分组   event -> event.getUserId().keyBy(OrderEvent::getUserId)// 设置窗口：10s，滚动窗口.window(TumblingEventTimeWindows.of(Time.seconds(10)))// 设置最大允许延迟时间.allowedLateness(Time.seconds(3))// 设置延迟很久数据侧边输出.sideOutputLateData(lateOutputTag)// 设置窗口函数，进行计算.apply(new OrderWindowFunction());// 4. 数据终端-sinkwindowStream.printToErr();// 获取侧边流中延迟数据DataStream<OrderEvent> lateOrderStream = windowStream.getSideOutput(lateOutputTag);lateOrderStream.printToErr("late>");// 5. 触发执行-executeenv.execute("StreamOrderWindowReport");}/*** 流式应用Checkpoint检查点设置*/private static void setEnvCheckpoint(StreamExecutionEnvironment env) {// 1. 启动Checkpointenv.enableCheckpointing(10000) ;// 2.设置StateBackendenv.setStateBackend(new HashMapStateBackend());// 3.设置Checkpoint存储env.getCheckpointConfig().setCheckpointStorage("file:///D:/ckpt/");// 4. 设置相邻Checkpoint至少时间间隔env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);// 5. 设置Checkpoint最大失败次数env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);// 6. 设置取消job时Checkpoint是删除还是保留env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);// 7.设置Checkpoint超时时间env.getCheckpointConfig().setCheckpointTimeout(10 * 60 * 1000);// 8. 设置Checkpoint最大并发次数env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);// 9. 设置模式env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);}