复盘笔记

专栏原创出处：github-源笔记文件 (opens new window) ，github-源码 (opens new window)，欢迎 Star，转载请附上原文出处链接和本声明。本节内容对应官方文档 (opens new window)，本节内容对应示例源码 (opens new window)

1 Time（时间）
2 Watermark（水印/水位线）
3 指定 Timestamp 与生成 Watermarks
- 3.1 SourceFunction 直接定义
- 3.2 通过 Flink 的 Timestamp Assigner 指定
4.为每个 Kafka 分区分配时间戳/水印

# 1 Time（时间）

所有由 Flink 事件-时间流应用生成的条目都必须伴随着一个时间戳。时间戳将一个条目与一个特定的时间点关联起来，一般这个时间点表示的是这条 record 发生的时间。不过 application 可以随意选择时间戳的含义，只要流中条目的时间戳是随着流的前进而递增即可。

支持的时间模型：

EventTime 是事件创建的时间，即事件产生时自带时间戳，在 Flink 处理计算中，事件时间难免有延迟，为了处理延迟，必须指定 Watermark 的生成方式
IngestionTime 是事件进入 Flink 的时间，即进入 source operator 时获取所在主机时间
ProcessingTime 是每一个算子操作的获取所在主机时间

时间模型比较

性能： ProcessingTime> IngestTime> EventTime
延迟： ProcessingTime< IngestTime< EventTime
确定性： EventTime> IngestTime> ProcessingTime

注意：Flink 从数据流模型中实现了许多技术。有关事件时间和水印的一个很好的介绍，请查看下

流式处理概念:时间域、窗口化 (opens new window)，中文译文 (opens new window)

流式处理概念:水印、触发器、积累模式 (opens new window)，中文译文 (opens new window)

流式处理概念:会话窗口 (opens new window), 中文译文 (opens new window)

如何设置时间域？

val env = StreamExecutionEnvironment.getExecutionEnvironment
// 默认使用 TimeCharacteristic.ProcessTime
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)

// 可选的:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

# 2 Watermark（水印/水位线）

实时系统中，由于各种原因造成的延时，造成某些消息发到 flink 的时间延时于事件产生的时间。
如果基于event time构建window，但是对于迟到的事件，我们又不能无限期的等下去，必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了。这个特别的机制，就是 Watermark。

Watermark 作为数据处理流中的一部分进行传输，并且携带一个时间戳 t。一个 Watermark(t) 表示流中应该不再有事件时间比 t 小的元素（某个事件的时间戳比 Watermark 时间大）

水印有两个基本属性：

它们必须单调递增，以确保任务的 event-time 时钟向前推进，而不是向后
它们与记录的时间戳是相关的。一个时间戳为 T 的水印表示的是：在它之后接下来的所有记录的时间戳，都必须大于 T

Watermarks(水印) 处理机制如下：

参考 google 的 DataFlow。
是 event time 处理进度的标志。
表示比 watermark 更早 (更老) 的事件都已经到达 (没有比水位线更低的数据 )。
基于 watermark 来进行窗口触发计算的判断。

# 2.1 有序流中 Watermarks

某些情况下，基于 Event Time 的数据流是有续的 (相对 event time)。
在有序流中，watermark 就是一个简单的周期性标记。

stream_watermark_in_order

# 2.2 乱序流中 Watermarks

在更多场景下，基于 Event Time 的数据流是无续的 (相对 event time)。

在无序流中，Watermarks 至关重要，他告诉 operator 比 Watermarks 更早 (更老/时间戳更小) 的事件已经到达，operator 可以将内部事件时间提前到 Watermarks 的时间戳 (可以触发 window 计算)

stream_watermark_out_of_order

# 2.3 并行流中的 Watermarks

通常情况下， watermark 在源函数中或源函数后生成。如果指定多次 watermark，后面指定的 watermark 会覆盖前面的值。源函数的每个 sub-task 独立生成水印。

随着水印在算子操作中的流动，它们会提前到达其到达的算子操作的事件时间。每当算子操作提前其事件时间时，同时算子操作会为下游生成一个新的 watermark。

一些算子消耗多个输入流；例如，keyBy(…) or partition(…) function。这样的算子的当前事件时间是其输入流的事件时间中的最小值。随着其输入流更新其事件时间，算子也将更新。

现在详细的解释一下，一个 task 如何释放一个水印，并在收到一个新的水印时如何更新它自身的 event-time 时钟（clock）。Flink 会将数据流分成不同的分区（partition），对于每个分区，都会有不同的 operator task 处理，这些 task 并行工作处理整个数据流。每个分区都是记录（包含时间戳）与水印的数据流。对于一个 operator，基于它与上游/下游 operators 连接的方式，它的 tasks 可以从一个或多个输入分区接受 records 和水印，并释放 records 和水印到一个或多个输出分区。下面我们会详细的介绍一个 task 如何释放水印到多个 output tasks，以及它如何根据（从输入 tasks）收到的水印，推进它自身的 event-time 时钟。

一个 task 对每个输入分区，都维护了一个分区水印。当 task 从一个分区收到一个水印，它会将对应分区的水印，更新为收到的水印最大值，并设置为当前值。然后，task 更新它的 event-time 时钟为所有分区水印中的最小值。如果 event-time 时钟相较之前有增加，则 task 处理所有被触发的计时器，并最终广播它的新事件-时间到所有下游 task，此操作通过释放一个对应的水印到所有连接的输出分区完成。

对于有多个输入流的（例如 Union 或 CoFlatMap 操作）operators，它们的 tasks 也会计算它们自身的 event-time 时钟，并作为所有分区水印的最小值– 他们并不（从不同的输入流中）区分 partition watermarks。这样做的结果是，两个不同的输入流中的数据会根据同一 event-time 时钟进行处理。但是，如果一个 application 的各个输入流的事件时间并不是一致的，则这个行为会导致问题。

Flink 的水印处理以及传播算法,确保了 operator task 恰当地释放一致时间戳的记录和水印。然而它依赖的基础是：所有分区持续提供递增的水印。一旦一个分区的水印不再递增，或者完全空闲（不再发送任何记录与水印），则 task 的事件-时间时钟不会再向前推进，并且 task 的计时器也不会被触发。在基于时间的、依赖于向前（advancing）时钟执行计算（并做清理）的 operators 中，便会造成问题。最终会导致处理延时、state 大小剧增（如果没有定期从所有的输入任务中接收到新的水印）。

若是两个输入流的水印差异太大，也会造成类似的影响。在有两个输入流的 task 中，它的事件-时钟会对应于较慢的流，并且较快的流的 records 或是中间结果一般会缓存到 state 中，直到 event-time 时钟允许处理它们。

parallel_streams_watermarks

# 3 指定 Timestamp 与生成 Watermarks

# 3.1 SourceFunction 直接定义

class GameSourceFunction[T <: GameModel](seq: Seq[T], millis: Long = 0) extends SourceFunction[T] {

  private var counter = 0
  private var isRunning = true

  override def run(ctx: SourceFunction.SourceContext[T]): Unit = {
    while (isRunning && counter < seq.length) {
      // ctx.collect(seq(counter))
      val next = seq(counter)
      ctx.collectWithTimestamp(next, next.eventTimestamp) // 毫秒时间戳
      //      if (next.hasWatermarkTime) {
      //        ctx.emitWatermark(new Watermark(next.getWatermarkTime))
      //      }
      counter = counter + 1
      Thread.sleep(millis)
    }
  }

  override def cancel(): Unit = {
    isRunning = false
  }
}

# 3.2 通过 Flink 的 Timestamp Assigner 指定

Flink 提供了两个接口用于指定 Timestamp 与 Watermarks

AssignerWithPeriodicWatermarks 按时间间隔周期性生成 Watermarks
AssignerWithPunctuatedWatermarks 根据接入的事件触发条件生成 Watermarks

生成类图关系如下： timestampAssigner_uml

DataStream支持指定Timestamp 与 Watermarks API

def assignTimestamps(extractor: TimestampExtractor[T]): DataStream[T] // 已废弃 @Deprecated
def assignAscendingTimestamps(extractor: T => Long): DataStream[T] // 底层转换为 AssignerWithPeriodicWatermarks
def assignTimestampsAndWatermarks(assigner: AssignerWithPeriodicWatermarks[T]): DataStream[T]
def assignTimestampsAndWatermarks(assigner: AssignerWithPunctuatedWatermarks[T]): DataStream[T]

简单示例：

sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 设置时间域
stream.assignTimestampsAndWatermarks(new GameAscendingTimestampExtractor[UserLogin]) // 设置水印生成器

####3.2.1 AssignerWithPeriodicWatermarks（周期性水印生成器）通过定义生成水印的间隔（每 n 毫秒） ExecutionConfig.setAutoWatermarkInterval(...)。
调用AssignerWithPeriodicWatermarks的getCurrentWatermark()方法，如果返回的水印非空且大于前一个水印，则覆盖以前的水印。

总结为：

基于 Timer
ExecutionConfig.setAutoWatermarkInterval(msec) (默认是 200ms, 设置 Watermarks 发送的周期)
实现 AssignerWithPeriodicWatermarks 接口

# 3.2.1.1 Flink-API 提供：时间戳单调递增的分配器

适用于 event-time 戳单调递增的场景，数据没有太多延时。

底层实现为 AscendingTimestampExtractor<T> implements AssignerWithPeriodicWatermarks<T>

val withTimestampsAndWatermarks = stream.assignAscendingTimestamps( _.getCreationTime )

# 3.2.1.2 Flink-API 提供：允许固定延迟的分配器

适用于预先知道最大延迟的场景 (例如最多比之前的元素延迟 3000ms)。

底层实现为 BoundedOutOfOrdernessTimestampExtractor<T> implements AssignerWithPeriodicWatermarks<T>

# 3.2.1.3 自定义实现 AssignerWithPeriodicWatermarks 示例


// 设置水印生成器
stream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator[UserLogin]())
stream.assignTimestampsAndWatermarks(new TimeLagWatermarkGenerator[UserLogin]())

/**
周期性水印生成器 = 示例 1
此生成器生成的水印支持处理给定延迟时间范围内的数据
支持延迟的时间动态计算 = 当前处理事件中的最大时间 - 支持最大延迟时间
  */
class BoundedOutOfOrdernessGenerator[T <: GameModel] extends AssignerWithPeriodicWatermarks[T] {

  val maxOutOfOrderness = 3500L // 支持最大延迟时间 3.5 seconds

  var currentMaxTimestamp: Long = _ // 当前最大时间

  override def extractTimestamp(element: T, previousElementTimestamp: Long): Long = {
    val timestamp = element.eventTimestamp
    currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
    timestamp
  }

  // 返回水印为当前最大时间减去支持最大延迟时间
  override def getCurrentWatermark: Watermark =
    new Watermark(currentMaxTimestamp - maxOutOfOrderness)
}

/**
周期性水印生成器 = 示例 2
此生成器生成的水印支持处理给定延迟时间范围内的数据。
支持延迟的时间动态计算 = 当前系统时间 - 支持最大延迟时间
  */
class TimeLagWatermarkGenerator[T <: GameModel] extends AssignerWithPeriodicWatermarks[T] {

  val maxTimeLag = 5000L // 支持最大延迟时间 5 seconds

  override def extractTimestamp(element: T, previousElementTimestamp: Long): Long =
    element.eventTimestamp

  // 返回水印为当前时间减去支持最大延迟时间
  override def getCurrentWatermark: Watermark =
    new Watermark(System.currentTimeMillis() - maxTimeLag)
}

# 3.2.2 AssignerWithPunctuatedWatermarks（条件水印生成器）

使用 AssignerWithPunctuatedWatermarks接口。
首先调用该 extractTimestamp(...)方法为元素分配时间戳，然后立即调用checkAndGetNextWatermark(...)方法。
如果返回的水印非空且大于前一个水印，则覆盖以前的水印。

总结为：

实现 AssignerWithPunctuatedWatermarks 接口
生成水印逻辑自定义

注意：可以在每个事件上生成水印。但是，由于每个水印都会在下游引起一些计算，因此过多的水印会降低性能。

// 设置水印生成器
stream.assignTimestampsAndWatermarks(new PunctuatedAssigner[UserLogin]())
    
/** 带条件的水印生成器 = 示例
在特定事件规则，可能会生成新的水印时生成水印
  */
class PunctuatedAssigner[T <: GameModel] extends AssignerWithPunctuatedWatermarks[T] {

  override def extractTimestamp(element: T, previousElementTimestamp: Long): Long = {
    element.eventTimestamp
  }

  override def checkAndGetNextWatermark(lastElement: T, extractedTimestamp: Long): Watermark = {
    if (lastElement.hasWatermarkMarker) new Watermark(extractedTimestamp) else null
  }
}

# 4.为每个 Kafka 分区分配时间戳/水印

当 kafka 作为数据源时，kafka 的每个 Partition 分区里面时间戳可能是升序或者乱序模式。通常情况，我们会多个 Partition 分区并行处理，我们可以为 kafka 配置水印。 kafka 内部为每个 Partition 分区维护一个水印，并且在流进行 shuffle 时以2.3 并行流中的 Watermarks进行水印合并

val kafkaSource = new FlinkKafkaConsumer09[MyType]("myTopic", schema, props)
kafkaSource.assignTimestampsAndWatermarks(new AscendingTimestampExtractor[MyType] {
    def extractAscendingTimestamp(element: MyType): Long = element.eventTimestamp
})

val stream: DataStream[MyType] = env.addSource(kafkaSource)

← DataStream概览 DataStream-ProcessFunction →