Flink Watermark和时间语义

Flink 中的时间语义

时间语义：

EventTime

：事件创建时间；

Ingestion Time

：数据进入

Flink

的时间；

Processing Time

：执行操作算子的本地系统时间，与机器无关。不同的时间语义有不同的应用场合，我们往往更关系事件时间

Event Time

。数据生成的时候就会自动注入时间戳，

Event Time

可以从日志数据的时间戳

timestamp)

中提取。

设置 Event Time

我们可以直接在代码中，对执行环境调用

setStreamTimeCharacteristic

方法，设置流的时间特性。具体的时间，还需要从数据中提取时间戳

timestamp

。

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

乱序数据的影响

当

Flink

以

Event Time

模式处理数据流时，它会根据数据里的时间戳来处理基于时间的算子。由于网络、分布式等原因，会导致乱序数据的产生。如上图所示，理想情况与实际情况会存在差异，乱序数据会让窗口计算不准确。解决方案是让窗口等几分钟。

水位线 Watermark

怎么避免乱序数据带来计算不正确？
遇到一个时间戳到达了窗口关闭时间，不应该立刻触发窗口计算，而是等待一段时间，等迟到的数据来了再关闭窗口。

Watermark

是一种衡量

Event Time

进展的机制，可以设置延迟触发。

Watermark

是用于处理乱序事件的，而正确的处理乱序事件，通常用

Watermark

机制结合

window

来实现。数据流中的

Watermark

用于表示

timestamp

小于

Watermark

的数据，都已经达到了，因此，

window

的执行也是由

Watermark

触发的。

Watermark

用来让程序自己延迟和结果正确性。

Watermark 的特点：

Watermark

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

Watermark

与数据的时间戳有关。

watermark 的传递、引入和设定

**

watermark

的传递：** 一个

Task

输入可以并行多个，如下有

4

个并行度，输出也可能存在多个并行，如下有3个。每个任务

Task

内部都有一个事件时钟，且每个分区也维护了对应的

WM

，如下的

Partition WM

。当事件流流进

Partition

时会判断新事件流的

WM

是否大于当前的

Partition WM

，当大于时就更新

Partition

的时间戳

WM

为新流入的

WM

（取最大值），如下

1->2

象限

Partition WM

的变化。同时，如下

Task

也维护了一个全局的

WM

表示事件时钟，该值取分区中最小的

WM

作为输出的时间戳，如下第二象限的输出选择最小的

WM=3

向下传递。当第二个(横线)分区

Partition WM

流进来

WM=7

的事件流时，就会出现第三象限的情景，但是最小的

WM

还是

=3

，因此不更新

Task

全局的

WM

。当第三个分区

Partition WM

流进来

WM=6

的事件流时，就会出现第四象限的情景，此时分区

Partition WM

的最小值

=4

，因此

Task

全局

WM

就

=4

。

**

watermark

的引入：**

Event Time

的使用一定要指定数据源中的时间戳。对于排好序的数据，只需要指定时间戳就够了，不需要延迟触发。

importorg.apache.flink.streaming.api.windowing.time.Time//同时分配时间戳和水位线
dataStream.assignTimestampsAndWatermarks(//无序数据       Time.milliseconds(1000)=延迟时间newBoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.milliseconds(1000)){//提取事件戳 = timestamp * 1000是因为出入的毫秒
  override def extractTimestamp(t:SensorReading):Long={
    t.timestamp *1000}})

【1】对于排好序的数据，不需要延迟触发，可以只指定事件戳就行了

dataStream.assignTimestampsAndWatermarks(_.timestamp *1000)

【2】

Flink

暴露了

TimestampAssigner

接口供我们实现，使我们可以自定义如何从事件数据中抽取时间戳和生成

watermark

。

MyAssigner

可以有两种类型，都继承自

TimestampAssigner

。

dataStream.assignTimestampsAndWatermarks(newMyAssigner())

**

TimestampAssigner

：定义了抽取时间戳，以及生成

watermark

的方法，有两种类型：**
**【1】

AssignerWithPeriodicWatermarks

：** 系统会周期性的将

Watermark

插入到流中。默认周期是

200

毫秒（如果是

processingTime

则

Watermark = 0

），可以使用

ExecutionConfig.setAutoWatermarkInterval()

方法进行设置。升序和前面乱序的处理

BoundedOutOfOrderness

，都是基于周期性

watermark

的。举例：如下产生

watermark

的逻辑：每隔

5

秒，

Flink

调用

AssignerWithPeriodicWatermarks

的

getCurrentWatermark()

方法。如果方法返回一个时间戳大于之前水位的时间戳，新的

water

会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于之前水位的时间戳，则不会产生新的

watermark

。

//方案一://EventTime是以数据自带的时间戳字段为准，应用程序需要指定如何从record中抽取时间戳字段
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)//每隔 5秒产生一个 watermark
env.getConfig.setAutoWatermarkInterval(5000);

​
//方案二//自定义一个周期性的时间戳classPeriodicAssignerextendsAssignerWithPeriodicWatermarks[SensorReading]{

  val bound:Long=60*1000//延时为 1 分钟var maxTs:Long=Long.MinValue//观察到的最大时间戳//生成水位线
  override def getCurrentWatermark:Watermark={newWatermark(maxTs - bound)}//抽取时间戳的方法
  override def extractTimestamp(t:SensorReading, l:Long):Long={
    maxTs = maxTs.max(t.timestamp)
    t.timestamp
  }}

**【2】

AssignerWithPunctuatedWatermarks

：** 没有时间周期规律，可打断的生成

watermark

。

classPunctuatedAssignerextendsAssignerWithPunctuatedWatermarks[SensorReading]{
  val bound:Long=60*1000//获取水位线，根据数据触发
  override def checkAndGetNextWatermark(t:SensorReading, l:Long):Watermark={if(t.id =="sensor_1"){newWatermark(l - bound)}else{null}}//抽取时间戳的方法
  override def extractTimestamp(t:SensorReading, l:Long):Long={
    t.timestamp
  }}

watermark 的设定：
【1】在

Flink

中，

watermark

由应用程序开发人员生成，这通常需要对相应的领域有一定的了解。
【2】如果

watermark

设置的延迟太久，收到结果的速度可能就会很慢，解决办法是在水位线到达之前输出一个近似结果。
【3】而如果

watermark

到达得太早，则可能收到错误结果，不过

Flink

处理迟到数据的机制可以解决这个问题。