Question

Flink 在流处理节目中支持不同的 时间 概念。

• 处理时间： 处理时间是指执行相应 算子操作的机器的系统时间。

  当流程序在处理时间运行时，所有基于时间的 算子操作（如时间窗口）将使用运行相应算子的机器的系统时钟。每小时处理时间窗口将包括在系统时钟指示整个小时之间到达特定算子的所有记录。例如，如果应用程序在上午 9:15 开始运行，则第一个每小时处理时间窗口将包括在上午 9:15 到上午 10:00 之间处理的事件，下一个窗口将包括在上午 10:00 到 11:00 之间处理的事件，因此上。

  处理时间是最简单的时间概念，不需要流和机器之间的协调。它提供最佳性能和最低延迟。但是，在分布式和异步环境中，处理时间不提供确定性，因为它容易受到记录到达系统的速度（例如从消息队列）到记录在系统内的算子之间流动的速度的影响。和停电（预定或其他）。

• 事件时间： 事件时间是每个事件在其生产设备上发生的时间。此时间通常在进入 Flink 之前嵌入记录中，并且 可以从每个记录中提取该 事件时间戳 。在事件时间，时间的进展取决于数据，而不是任何挂钟。事件时间程序必须指定如何生成 事件时间水印 ，这是表示事件时间进度的机制。该水印机制在 下面 的后面部分中描述。

  在一个完美的世界中，事件时间处理将产生完全一致和确定的结果，无论事件何时到达，或者它们的排序。但是，除非事件已知按顺序到达（按时间戳），否则事件时间处理会在等待无序事件时产生一些延迟。由于只能等待一段有限的时间，因此限制了确定性事件时间应用程序的可能性。

  假设所有数据都已到达，事件时间 算子操作将按预期运行，即使在处理无序或延迟事件或重新处理历史数据时也会产生正确且一致的结果。例如，每小时事件时间窗口将包含带有落入该小时的事件时间戳的所有记录，无论它们到达的顺序如何，或者何时处理它们。（有关更多信息，请参阅有关 迟发事件 的部分。）

  请注意，有时当事件时间程序实时处理实时数据时，它们将使用一些 处理时间 算子操作，以确保它们及时进行。

• 摄取时间： 摄取时间是事件进入 Flink 的时间。在源算子处，每个记录将源的当前时间作为时间戳，并且基于时间的 算子操作（如时间窗口）引用该时间戳。

  摄取时间 在概念上位于 事件时间 和 处理时间之间 。与 处理时间 相比 ，它稍贵一些，但可以提供更可预测的结果。因为 摄取时间 使用稳定的时间戳（在源处分配一次），所以对记录的不同窗口 算子操作将引用相同的时间戳，而在 处理时间中， 每个窗口算子可以将记录分配给不同的窗口（基于本地系统时钟和任何运输延误）。

  与 事件时间 相比， 摄取时间 程序无法处理任何无序事件或后期数据，但程序不必指定如何生成 水印 。

  在内部， 摄取时间 与 事件时间 非常相似，但具有自动时间戳分配和自动水印生成函数。

设定时间特征

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

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

• Java
• Scala

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

// alternatively:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));

stream
  .keyBy( (event) -> event.getUser() )
  .timeWindow(Time.hours(1))
  .reduce( (a, b) -> a.add(b) )
  .addSink(...);

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)

// alternatively:
// 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(...)

请注意，为了在 事件时间 运行此示例，程序需要使用直接为数据定义事件时间的源并自行发出水印，或者程序必须在源之后注入 时间戳分配器和水印生成器 。这些函数描述了如何访问事件时间戳，以及事件流表现出的无序程度。

以下部分描述了 时间戳 和 水印 背后的一般机制。有关如何在 Flink DataStream API 中使用时间戳分配和水印生成的指南，请参阅 生成时间戳/水印 。