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 的使用一定要指定数据源中的时间戳。对于排好序的数据，只需要指定时间戳就够了，不需要延迟触发。

import org.apache.flink.streaming.api.windowing.time.Time
//同时分配时间戳和水位线
dataStream.assignTimestampsAndWatermarks(
//无序数据       Time.milliseconds(1000)=延迟时间
new BoundedOutOfOrdernessTimestampExtractor[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(new MyAssigner())

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);

?
//方案二
//自定义一个周期性的时间戳
class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading]{

  val bound: Long = 60 * 1000 //延时为 1 分钟
  var maxTs: Long = Long.MinValue //观察到的最大时间戳

  //生成水位线
  override def getCurrentWatermark: Watermark = {
    new Watermark(maxTs - bound)
  }

  //抽取时间戳的方法
  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    maxTs = maxTs.max(t.timestamp)
    t.timestamp
  }
}

【2】AssignerWithPunctuatedWatermarks：没有时间周期规律，可打断的生成 watermark。

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading]{
  val bound: Long = 60 * 1000

  //获取水位线，根据数据触发
  override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark = {
    if(t.id == "sensor_1"){
      new Watermark(l - bound)
    }else{
      null
    }
  }

  //抽取时间戳的方法
  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    t.timestamp
  }
}

watermark 的设定：
【1】在 Flink 中，watermark 由应用程序开发人员生成，这通常需要对相应的领域有一定的了解。
【2】如果 watermark 设置的延迟太久，收到结果的速度可能就会很慢，解决办法是在水位线到达之前输出一个近似结果。
【3】而如果 watermark 到达得太早，则可能收到错误结果，不过 Flink 处理迟到数据的机制可以解决这个问题。