Flink DataStream 时间语义

Event Time - Processing Time - Ingestion Time

Flink 在流式传输程序中支持不同的_时间_概念。

• 处理时间(Processing Time)：简单的理解，就是数据在窗口中，被处理执行的时间。下面是官方的释义：处理时间是指执行相应操作的机器的系统时间。当流式程序按处理时间运行时，所有基于时间的操作（如时间窗口）都将使用运行相应操作员的计算机的系统时钟。每小时处理时间窗口将包括系统时钟指示整小时的时间之间到达特定操作员的所有记录。例如，如果应用程序在9:15 am开始运行，则第一个每小时处理时间窗口将包括在9:15 am和10:00 am之间处理的事件，下一个窗口将包括在10:00 am和11:00 am之间处理的事件，依此类推。处理时间是最简单的时间概念，不需要流和机器之间的协调。它提供了最佳的性能和最低的延迟。但是，在分布式和异步环境中，处理时间不能提供确定性，因为它容易受到记录到达系统（例如从消息队列）到达系统的速度，记录在系统内部操作员之间流动的速度的影响。 以及中断（计划的或其他方式）。

• 事件时间(Event time)：简单的理解，就是数据在产生的时间，例如用户A在x时间，点击了B商品，此时所产生的数据的时间时间就是x。下面是官方释义：事件时间是每个事件在其生产设备上发生的时间。该时间通常在它们进入Flink之前嵌入到记录中，并且 可以从每个记录中提取_事件时间戳_。在事件时间中，时间的进度取决于数据，而不取决于任何挂钟。事件时间程序必须指定如何生成“ _事件时间水印”_，这是一种表示事件时间进展的机制。在理想情况下，事件时间处理将产生完全一致且确定的结果，而不管事件何时到达或它们的顺序如何。但是，除非已知事件是按时间戳（按时间戳）到达的，否则事件时间处理会在等待无序事件时产生一定的延迟。由于只能等待有限的时间，因此这限制了确定性事件时间应用程序的可用性。
  假设所有数据都已到达，事件时间操作将按预期方式运行，即使在处理无序或迟到的事件或重新处理历史数据时，也会产生正确且一致的结果。例如，每小时事件时间窗口将包含所有带有落入该小时事件时间戳的记录，无论它们到达的顺序或处理的时间。
  请注意，有时当事件时间程序实时处理实时数据时，它们将使用一些_处理时间_操作，以确保它们及时进行。

• 摄入时间(Ingestion Time)：摄取时间是事件进入Flink的时间。在源操作员处，每个记录都将源的当前时间作为时间戳记，并且基于时间的操作（如时间窗口）引用该时间戳记。
  _摄取时间_从概念上讲介于_事件时间和处理时间之间_。与_处理时间_相比 ，它稍微贵一些，但结果却更可预测。由于 _摄取时间_使用稳定的时间戳（在源处分配了一次），因此对记录的不同窗口操作将引用相同的时间戳，而在_处理时间中，_每个窗口操作员都可以将记录分配给不同的窗口（基于本地系统时钟和任何运输延误）。
  与_事件时间_相比，_提取时间_程序无法处理任何乱序事件或迟到的数据，但是程序不必指定如何生成_水印_。
  在内部，_摄取时间与事件时间_非常相似，但是具有自动时间戳分配和自动水印生成功能。
  

在实际工作中，Ingestion Time使用的场景较少。并且是不设置时间语义是，默认使用process time

设置时间特征（Setting a Time Characteristic）

Flink DataStream程序的第一部分通常设置基准_时间特征_。该设置定义了数据流源的行为方式（例如，是否分配时间戳），以及诸如的窗口操作应使用什么时间概念KeyedStream.timeWindow(Time.seconds(30))。


以下示例显示了一个Flink程序，该程序在每小时的时间窗口中汇总事件。窗口的行为与时间特征相适应。

val env = StreamExecutionEnvironment.getExecutionEnvironment

//默认为process time，所以如果选择ProcessingTime，也可以不进行此步的设置
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val stream: DataStream[MyEvent] = env.addSource(new FlinkKafkaConsumer09[MyEvent](topic, schema, props))
stream
    .keyBy( _.getUser )
    .timeWindow(Time.hours(1))
    .reduce( (a, b) => a.add(b) )
    .addSink(...)

请注意，为了在事件时间中运行此示例，程序需要使用直接为数据定义事件时间并自己发出水印的源，或者程序必须在源之后注入Timestamp Assigner＆Watermark Generator。这些功能描述了如何访问事件时间戳，以及事件流呈现出何种程度的乱序。


现实世界中的时间是不一致的，在 flink 中被划分为事件时间，摄入时间，处理时间三种。如果以 EventTime 为基准来定义时间窗口那将形成 EventTimeWindow,*要求消息本身就应该携带 EventTime *


如果以 IngesingtTime 为基准来定义时间窗口那将形成 IngestingTimeWindow,以 source 的 systemTime 为准。 如果以 ProcessingTime 基准来定义时间窗口那将形成 ProcessingTimeWindow，以 operator 的 systemTime 为准。


以上三种选择是需要根据不同业务需求进行选择的，例如，在数据计算的过程中，需要考虑到网络延迟的原因，那么就应该选择setStreamTimeCharacteristic(TimeCharacteristic.EventTime)，相反，不需要考虑延迟原因的话，只考虑处理数据的时间，则可以选择另两个。