Flink中的时间语义

• Event Time：事件创建时间；
• Ingestion Time：数据进入Flink的时间；
• Processing Time：执行操作算子的本地系统时间，与机器相关；

Event Time是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。

Flink-时间语义与Wartmark及EventTime在Window中的使用

• 不同的时间语义有不同的应用场合
• 我们往往更关心事件事件（Event Time）

​ 这里假设玩游戏，两分钟内如果过5关就有奖励。用户坐地铁玩游戏，进入隧道前已经过3关，在隧道中又过了2关。但是信号不好，后两关通关的信息，等到出隧道的时候（8:23:20）才正式到达服务器。

​ 如果为了用户体验，那么应该按照Event Time处理信息，保证用户获得游戏奖励。

• Event Time可以从日志数据的时间戳（timestamp）中提取2017-11-02 18:27:15.624 INFO Fail over to rm

EventTime的引入

​ 在Flink的流式处理中，绝大部分的业务都会使用eventTime，一般只在eventTime无法使用时，才会被迫使用ProcessingTime或者IngestionTime。

​ （虽然默认环境里使用的就是ProcessingTime，使用EventTime需要另外设置）

​ 如果要使用EventTime，那么需要引入EventTime的时间属性，引入方式如下所示：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 从调用时刻开始给env创建的每一个stream追加时间特征
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

注：具体的时间，还需要从数据中提取时间戳。

Watermark

概念

• Flink对于迟到数据有三层保障，先来后到的保障顺序是：WaterMark => 约等于放宽窗口标准allowedLateness => 允许迟到（ProcessingTime超时，但是EventTime没超时）sideOutputLateData => 超过迟到时间，另外捕获，之后可以自己批处理合并先前的数据

Flink-时间语义与Wartmark及EventTime在Window中的使用

​ 我们知道，流处理从事件产生，到流经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。

​ Watermark 就是触发前一窗口的“关窗时间”，一旦触发关门那么以当前时刻为准在窗口范围内的所有所有数据都会收入窗中。

​ 只要没有达到水位那么不管现实中的时间推进了多久都不会触发关窗。

Watermark的特点

Flink-时间语义与Wartmark及EventTime在Window中的使用

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

Watermark的传递

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

Watermark的引入

​ watermark的引入很简单，对于乱序数据，最常见的引用方式如下：

dataStream.assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor(Time.milliseconds(1000)) {
  @Override
  public long extractTimestamp(element: SensorReading): Long = { 
    return element.getTimestamp() * 1000L;
  } 
});

​ **Event Time的使用一定要指定数据源中的时间戳。否则程序无法知道事件的事件时间是什么(数据源里的数据没有时间戳的话，就只能使用Processing Time了)**。

​ 我们看到上面的例子中创建了一个看起来有点复杂的类，这个类实现的其实就是分配时间戳的接口。Flink暴露了TimestampAssigner接口供我们实现，使我们可以自定义如何从事件数据中抽取时间戳。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置事件时间语义 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream dataStream = env.addSource(new SensorSource()) .assignTimestampsAndWatermarks(new MyAssigner());

MyAssigner有两种类型

• AssignerWithPeriodicWatermarks
• AssignerWithPunctuatedWatermarks

以上两个接口都继承自TimestampAssigner。

TimestampAssigner

AssignerWithPeriodicWatermarks

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

AssignerWithPunctuatedWatermarks

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