Flink Event Time / Processing Time / Ingestion Time （事件时间/处理时间/摄取时间）

╰半橙微兮° 2021-09-25 02:16 362阅读 0赞

### 概念 ###

Flink supports different notions of *time* in streaming programs.  
flink支持不同的 *time* 流媒体的概念。

*  **处理时间：** 处理时间是指正在执行相应操作的机器的系统时间。  
    当流式程序在处理时间上运行时，所有基于时间的操作(如时间窗口)都将使用运行相应操作员的机器的系统时钟。每小时处理时间窗口将包括在系统时钟指示整个小时的时间之间到达特定操作员的所有记录。例如，如果应用程序在上午9：15开始运行，则第一个每小时处理时间窗口将包括上午9：15至10：00之间处理的事件，下一个窗口将包括上午10：00至11：00之间处理的事件，依此类推。  
    处理时间是最简单的时间概念，并且不需要流和机器之间的协调。它提供了最佳性能和最低延迟。但是，在分布式和异步环境中，处理时间并不提供决定论，因为它容易受到记录到达系统的速度（例如，从消息队列）到系统内部操作员之间记录流的速度，以及停机（预定或其他）。

*  **事件时间：** 事件时间是每个事件发生在其生产设备上的时间。此时间通常在记录进入Flink之前嵌入其中，并且可以从每个记录中提取该 *event timestamp* 。在事件时间，时间的进展取决于数据，而不是任何墙上的时钟。事件时程序必须指定如何生成 *事件时间水印*，这是一种在事件时间内表示进度的机制。这种水印机制将在后面的一节中描述，[下面][Link 1]。  
    在一个完美的世界中，事件时间处理将产生完全一致和决定性的结果，而不管事件何时到达，或事件的顺序如何。但是，除非已知事件按顺序到达(按时间戳)，否则事件时间处理在等待无序事件时会产生一些延迟。由于只能等待一段有限的时间，这就限制了确定性事件时间应用程序的性能。  
    假设所有数据已经到达，事件时间操作将按预期运行，即使在与无序或延迟事件一起工作时或在再处理历史数据时也会产生正确且一致的结果。例如，每小时事件时间窗口将包含所有记录，该记录携带一个落入该小时的事件时间戳，而不管它们到达的顺序，还是在它们被处理时。（有关详细信息，请参见\[晚期事件\]（\#延迟元件）上的部分。)  
    注意，有时当事件时间程序正在实时地处理实时数据时，它们将使用一些 *processing time*  操作，以保证它们以及时的方式进行。

*  **摄入时间：** 摄入时间是事件进入Flink的时间。在源操作符中，每个记录以时间戳的形式获取源的当前时间，而基于时间的操作(如时间窗口)引用该时间戳。  
    ***Ingestion time*  概念上位于 *event time* 和 *processing time* 之间。**与 *processing time* 相比，它的成本略高，但结果更可预测。由于摄取时间*使用稳定的时间戳(在源上分配一次)，对记录的不同窗口操作将引用相同的时间戳，而* processing time\_ 窗口运算符可以将记录分配给不同的窗口(基于本地系统时钟和任何传输延迟)。  
    与 *event time* 相比，*ingestion time* 程序无法处理任何无序事件或后期数据，但程序不必指定如何生成 *watermarks*。  
    内部，*ingestion time* 被处理得像 *event time*，但具有自动时间戳分配和自动水印生成。

## ##

### Setting a Time Characteristic（设定时间特征） ###

Flink Datastream程序的第一部分通常设置基本 *time characteristic*。该设置定义了数据流源的行为方式(例如，它们是否将分配时间戳)，以及诸如‘KeyedStream.timeWindow(Time.秒(30)’)之类的窗口操作应该使用什么时间概念。

以下示例显示了在每小时时间窗口中聚合事件的FLink程序。Windows的行为随时间特性而变化。

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

如果不设置，StreamExecutionEnvironment则默认是process_time!

flink 1.9.2

[Link 1]: https://www.yuque.com/apachecn/flink-doc-zh/c310dbad340350eafceb010136d17512