11.1 Flink中的时间语义

• Event Time：事件创建时间；

• Ingestion Time：数据进入Flink的时间；

• Processing Time：执行操作算子的本地系统时间，与机器相关；

• 不同的时间语义有不同的应用场合

• 我们往往更关心事件事件（Event Time）

• Event Time可以从日志数据的时间戳（timestamp）中提取

11.2 EventTime的引入

在Flink的流式处理中，绝大部分的业务都会使用eventTime，一般只在eventTime无法使用时，才会被迫使用ProcessingTime或者IngestionTime。
 （虽然默认环境里使用的就是ProcessingTime，使用EventTime需要另外设置）
如果要使用EventTime，那么需要引入EventTime的时间属性，引入方式如下所示：
注：具体的时间，还需要从数据中提取时间戳。

11.3 Watermark

Flink流计算编程--watermark（水位线）简介

11.3.1 概念

• Flink对于迟到数据有三层保障，先来后到的保障顺序是：
• WaterMark => 约等于放宽窗口标准
• allowedLateness => 允许迟到（ProcessingTime超时，但是EventTime没超时）
• sideOutputLateData => 超过迟到时间，另外捕获，之后可以自己批处理合并先前的数据
我们知道，流处理从事件产生，到流经source，再到operator，中间是有一个过程和时间的，虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、分布式等原因，导致乱序的产生，所谓乱序，就是指Flink接收到的事件的先后顺序不是严格按照事件的Event Time顺序排列的。





那么此时出现一个问题，一旦出现乱序，如果只根据eventTime决定window的运行，我们不能明确数据是否全部到位，但又不能无限期的等下去，此时必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了，这个特别的机制，就是Watermark。
• 当Flink以Event Time模式处理数据流时，它会根据数据里的时间戳来处理基于时间的算子。
（比如5s一个窗口，那么理想情况下，遇到时间戳是5s的数据时，就认为[0,5s)时间段的桶bucket就可以关闭了。）
• 实际由于网络、分布式传输处理等原因，会导致乱序数据的产生
• 乱序数据会导致窗口计算不准确
（如果按照前面说法，获取到5s时间戳的数据，但是2s，3s乱序数据还没到，理论上不应该关闭桶）
• 怎样避免乱序数据带来的计算不正确？
• 遇到一个时间戳达到了窗口关闭时间，不应该立即触发窗口计算，而是等待一段时间，等迟到的数据来了再关闭窗口
1. Watermark是一种衡量Event Time进展的机制，可以设定延迟触发
2. Watermark是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark机制结合window来实现
3. 数据流中的Watermark用于表示”timestamp小于Watermark的数据，都已经到达了“，因此，window的执行也是由Watermark触发的。
4. Watermark可以理解成一个延迟触发机制，我们可以设置Watermark的延时时长t，每次系统会校验已经到达的数据中最大的maxEventTime，然后认定eventTime小于maxEventTime - t的所有数据都已经到达，如果有窗口的停止时间等于maxEventTime – t，那么这个窗口被触发执行。
Watermark = maxEventTime-延迟时间t
5. watermark 用来让程序自己平衡延迟和结果正确性
watermark可以理解为把原本的窗口标准稍微放宽了一点。（比如原本5s，设置延迟时间=2s，那么实际等到7s的数据到达时，才认为是[0,5）的桶需要关闭了）
有序流的Watermarker如下图所示：（延迟时间设置为0s）
此时以5s一个窗口，那么EventTime=5s的元素到达时，关闭第一个窗口，下图即W(5)，W(10)同理。



乱序流的Watermarker如下图所示：（延迟时间设置为2s）
乱序流，所以可能出现EventTime前后顺序不一致的情况，这里延迟时间设置2s，第一个窗口则为5s+2s，当EventTime=7s的数据到达时，关闭第一个窗口。第二个窗口则是5*2+2=12s，当12s这个EventTime的数据到达时，关闭第二个窗口。



当Flink接收到数据时，会按照一定的规则去生成Watermark，这条Watermark就等于当前所有到达数据中的maxEventTime-延迟时长，也就是说，Watermark是基于数据携带的时间戳生成的，一旦Watermark比当前未触发的窗口的停止时间要晚，那么就会触发相应窗口的执行。
 由于event time是由数据携带的，因此，如果运行过程中无法获取新的数据，那么没有被触发的窗口将永远都不被触发。
上图中，我们设置的允许最大延迟到达时间为2s，所以时间戳为7s的事件对应的Watermark是5s，时间戳为12s的事件的Watermark是10s，如果我们的窗口1是1s~5s，窗口2是6s~10s，那么时间戳为7s的事件到达时的Watermarker恰好触发窗口1，时间戳为12s的事件到达时的Watermark恰好触发窗口2。

11.3.2watermark特点



• watermark 是一条特殊的数据记录
• watermark 必须单调递增，以确保任务的事件时间时钟在向前推进，而不是在后退
• watermark 与数据的时间戳相关

11.3.3Watermark的传递



1. 图一，当前Task有四个上游Task给自己传输WaterMark信息，通过比较，只取当前最小值作为自己的本地Event-time clock，上图中，当前Task[0,2)的桶就可关闭了，因为所有上游中2s最小，能保证2s的WaterMark是准确的（所有上游Watermark都已经>=2s)。这时候将Watermark=2广播到当前Task的下游。
2. 图二，上游的Watermark持续变动，此时Watermark=3成为新的最小值，更新本地Task的event-time clock，同时将最新的Watermark=3广播到下游
3. 图三，上游的Watermark虽然更新了，但是当前最小值还是3，所以不更新event-time clock，也不需要广播到下游
4. 图四，和图二同理，更新本地event-time clock，同时向下游广播最新的Watermark=4

11.3.4 Watermark的引入

watermark的引入很简单，对于乱序数据，最常见的引用方式如下：
**Event Time的使用一定要指定数据源中的时间戳。否则程序无法知道事件的事件时间是什么(数据源里的数据没有时间戳的话，就只能使用Processing Time了)**。
我们看到上面的例子中创建了一个看起来有点复杂的类，这个类实现的其实就是分配时间戳的接口。Flink暴露了TimestampAssigner接口供我们实现，使我们可以自定义如何从事件数据中抽取时间戳。
MyAssigner有两种类型
• AssignerWithPeriodicWatermarks
• AssignerWithPunctuatedWatermarks
以上两个接口都继承自TimestampAssigner。

TimestampAssigner

AssignerWithPeriodicWatermarks

• 周期性的生成 watermark：系统会周期性的将 watermark 插入到流中
• 默认周期是200毫秒，可以使用 ExecutionConfig.setAutoWatermarkInterval() 方法进行设置
• 升序和前面乱序的处理 BoundedOutOfOrderness ，都是基于周期性 watermark 的。

AssignerWithPunctuatedWatermarks

• 没有时间周期规律，可打断的生成 watermark（即可实现每次获取数据都更新watermark）

11.3.5 Watermark的设定

• 在Flink中，Watermark由应用程序开发人员生成，这通常需要对相应的领域有一定的了解
• 如果Watermark设置的延迟太久，收到结果的速度可能就会很慢，解决办法是在水位线到达之前输出一个近似结果
• 如果Watermark到达得太早，则可能收到错误结果，不过Flink处理迟到数据的机制可以解决这个问题
 一般大数据场景都是考虑高并发情况，所以一般使用周期性生成Watermark的方式，避免频繁地生成Watermark。
注：一般认为Watermark的设置代码，在里Source步骤越近的地方越合适。

11.3.6 测试代码

测试Watermark和迟到数据
这里设置的Watermark的延时时间是2s，实际一般设置和window大小一致。
启动本地socket，输入数据，查看结果
输入：

分析

1. 计算窗口起始位置Start和结束位置End
从TumblingProcessingTimeWindows类里的assignWindows方法，我们可以得知窗口的起点计算方法如下： $$ 窗口起点start = timestamp - (timestamp -offset+WindowSize) % WindowSize $$ 由于我们没有设置offset，所以这里start=第一个数据的时间戳1547718199-(1547718199-0+15)%15=1547718195
计算得到窗口初始位置为Start = 1547718195，那么这个窗口理论上本应该在1547718195+15的位置关闭，也就是End=1547718210
计算修正后的Window输出结果的时间
测试代码中Watermark设置的maxOutOfOrderness最大乱序程度是2s，所以实际获取到End+2s的时间戳数据时（达到Watermark），才认为Window需要输出计算的结果（不关闭，因为设置了允许迟到1min）
所以实际应该是1547718212的数据到来时才触发Window输出计算结果。

窗口起始点和偏移量

flink-Window Assingers(窗口分配器)中offset偏移量
时间偏移一个很大的用处是用来调准非0时区的窗口，例如:在中国你需要指定一个8小时的时间偏移。