Flink学习笔记（四）：Flink 四大基石之 Window 和 Time

文章目录

1、概述
2、 Flink 的 Window 和 Time
- 2.1、Window API
- - 2.1.1、WindowAssigner
  - 2.1.2、Trigger
  - 2.1.3、Evictor
- 2.2、窗口类型
- - 2.2.1、Tumbling Windows
  - 2.2.2、Sliding Windows
  - 2.2.3、Session Windows
  - 2.2.4、Global Windows
- 2.3、Time 时间语义
- 2.4、乱序和延迟数据处理
- 2.5、综合案例

1、概述

Flink 基石

窗口 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>>() {
				@Override
				public 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") ;

				@Override
				public 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>>() {
				@Override
				public 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") ;

				@Override
				public 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）按活动会话对元素进行分组。与滚动窗口和滑动窗口相比，会话窗口不重叠，也没有固定的开始和结束时间。相反，当会话窗口在一段时间内未收到元素时（即，当出现不活动间隙时），会话窗口将关闭。会话窗口分配器可以配置静态会话间隙或会话间隙提取器功能，该函数定义不活动时间的时间。当此时间段到期时，当前会话将关闭，后续元素将分配给新的会话窗口。
Session Windows
示例代码：

// 3-1. 过滤和转换数据类型
		SingleOutputStreamOperator<Integer> mapStream = inputStream
			.filter(line -> line.trim().length() > 0)
			.map(new MapFunction<String, Integer>() {
				@Override
				public 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") ;

				@Override
				public 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>() {
				@Override
				public 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>() {
				@Override
				public 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
							@Override
							public long extractTimestamp(String element, long recordTimestamp) {
								// 2022-04-01 09:00:01,a,1 -> 2022-04-01 09:00:01 -> 1648774801000
								System.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. 执行环境-env
		Configuration configuration = new Configuration();
		configuration.setString("rest.port", "8081");
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
		env.setParallelism(1) ;
		// todo: 设置Checkpoint
		setEnvCheckpoint(env) ;
		// todo: 设置重启策略
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));

		// 2. 数据源-source
		DataStreamSource<String> inputStream = env.socketTextStream("127.0.0.1", 9999);

		// 3. 数据转换-transformation
		/*
			业务数据：
				o_101,u_121,11.50,2022-04-05 10:00:02
			3-1. 过滤、解析和封装数据
			3-2. 设置事假时间字段值和水位线Watermark
			3-3. 窗口设置及处理数据
		 */
		// 3-1. 过滤、解析和封装数据
		SingleOutputStreamOperator<OrderEvent> orderStream = inputStream
			.filter(line -> null != line && line.trim().split(",").length == 4)
			.map(new MapFunction<String, OrderEvent>() {
				@Override
				public 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. 设置事假时间字段值和水位线Watermark
		SingleOutputStreamOperator<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
					@Override
					public 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. 数据终端-sink
		windowStream.printToErr();

		// 获取侧边流中延迟数据
		DataStream<OrderEvent> lateOrderStream = windowStream.getSideOutput(lateOutputTag);
		lateOrderStream.printToErr("late>");

		// 5. 触发执行-execute
		env.execute("StreamOrderWindowReport");
	}

	/**
	 * 流式应用Checkpoint检查点设置
	 */
	private static void setEnvCheckpoint(StreamExecutionEnvironment env) {
		// 1. 启动Checkpoint
		env.enableCheckpointing(10000) ;

		// 2.设置StateBackend
		env.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);
	}