2.1、Flink流处理（Stream API）- Event Time（事件时间）之基础概览

妖狐艹你老母 2023-06-07 12:24 144阅读 0赞

**Table of Contents**

Event Time / Processing Time / Ingestion Time

设定时间特性

Event Time and Watermarks

Watermarks in Parallel Streams（平行流中的水印）

Late Elements（末元素）

Idling sources（闲置资源）

调试水印

算子如何处理水印

--------------------

# Event Time / Processing Time / Ingestion Time #

Flink 支持流程序中不同的时间概念。

*  **Processing time：**执行数据处理的机器的系统时间。

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

processing time 是最简单的时间概念，不需要流和机器之间的协调。它提供了最好的性能和最低的延迟。然而，在分布式和异步环境中 processing time 不提供准确性保证，因为它容易受到记录到达系统的速度的影响（例如，从消息队列来的数据）记录在系统内部算子之间流动的速度，以及中断的速度。

*  **Event time：**该事件发生时的时间。这个时间通常是在记录进入Flink之前嵌入的，可以从每个记录中提取事件时间戳。在事件时间，时间的进展取决于数据，而不是任何挂钟。事件时间程序必须指定如何生成 *Event Time Watermarks*，这是事件时间进展的信号机制。这是在事件时间上发出进展信号的机制。

事件时间处理将产生完全一致和确定的结果，不管事件何时到达，也不管它们的顺序如何。但是，除非知道事件是按顺序到达的(通过时间戳)，否则在等待无序事件时，事件时间处理会导致一些延迟。由于只能等待有限的一段时间，这就限制了应用程序的确定性事件时间。

假设所有数据都已到达，事件时间操作将按照预期的方式运行，即使在处理无序或延迟的事件或重新处理历史数据时，也会产生正确和一致的结果。例如，每小时事件时间窗口将包含所有记录，这些记录携带属于该小时的事件时间戳，而与它们到达的顺序无关，也与它们被处理的时间无关。

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

*  **Ingestion time：**摄入时间是事件进入 Flink 的时间。在源操作符中，每个记录以时间戳的形式获取源的当前时间，基于时间的操作(如时间窗口)引用该时间戳。

摄入时间概念上介于 event time 和 processing time 之间。与 processing time 相比，它稍微昂贵一些，但是提供了更可预测的结果。由于摄取时间使用稳定的时间戳(在源处分配一次)，对记录的不同窗口操作将引用相同的时间戳，而在处理时间中，每个窗口算子可以将记录分配到不同的窗口(基于本地系统时钟和任何传输延迟)。

与 event time 相比，摄取时间程序不能处理任何无序的事件或延迟的数据，但程序不必指定如何生成水印。

在内部，ingestion time 处理得很像 event time，但具有自动时间戳分配和自动水印生成。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70][]

## 设定时间特性 ##

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

下面的示例显示了一个按小时时间窗口聚合事件的 Flink 程序。窗口的行为适应了时代的特征。

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

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

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

# Event Time and Watermarks #

注意：Flink 实现了来自数据流模型的许多技术。有关事件时间和水印的详细介绍，请参阅下面的文章。

支持 event time 的流处理器需要一种方法来度量 event time 的进度。例如，当 event time 超过一小时结束时，需要通知构建每小时窗口的窗口算子，以便算子可以关闭正在运行的窗口。

Event time 可以独立于 Processing time。例如，在一个程序中，算子的当前 event time 可能稍微落后于 processing time(考虑到接收事件的延迟)，而两者的处理速度相同。另一方面，另一个流媒体程序可能只需要几秒钟的处理时间，就可以通过快进的方式处理Kafka主题(或另一个消息队列)中已经缓存的一些历史数据。

--------------------

在 Flink 中测量 event time 进展的机制是 watermarks。Watermarks 作为数据流的一部分，带有时间戳 t。水印(t)声明 event time 已经到达该流中的时间t，这意味着在时间戳t ' <= t的流中不应该有更多的元素，即时间戳较旧或等于水印的事件。

下图显示了带有(逻辑)时间戳的事件流，以及内联的水印。在这个例子中，事件是按顺序排列的(相对于它们的时间戳)，这意味着水印只是流中的周期标记。

![20191022162853607.png][]

对于无序流，Watermarks 是至关重要的，如下所示，其中事件不是按照它们的时间戳排序的。一般来说，水印是一种声明，在流中的那个点之前，在某个时间戳之前的所有事件都应该到达。当水印到达算子时，算子可以将其内部事件时间时钟提前到水印的值。

![20191022163037991.png][]

请注意，事件时间由新创建的流元素(或多个元素)继承自产生它们的事件或触发这些元素创建的水印。

## Watermarks in Parallel Streams（平行流中的水印） ##

水印是在源函数处或直接在源函数之后生成的。源函数的每个并行子任务通常独立地生成其水印。这些水印定义了特定并行源的事件时间。

当这些水印流经流媒体程序时，它们会将事件时间提前到它们到达的算子处。每当一个算子提前它的事件时间，它为它的后继算子产生一个新的下游水印。

一些算子消耗多个输入流;例如，一个union，或者一个keyBy(…)或partition(…)函数后面的算子。此类算子的当前事件时间是其输入流的事件时间的最小值。当它的输入流更新它们的事件时间时，算子也更新它们的事件时间。

下图显示了事件和水印在并行流中流动的示例，以及算子跟踪事件时间的示例。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 1][]

注意：Kafka源代码支持每个分区的水印。

## Late Elements（末元素） ##

某些元素可能会违反水印条件，这意味着即使在水印(t)出现之后，仍然会出现更多具有时间戳t ' <= t的元素。实际上，在许多实际的设置中，某些元素可以任意延迟，因此不可能指定某个事件时间戳的所有元素发生的时间。此外，即使延迟可以被限制，延迟太多的水印通常也是不可取的，因为它会导致事件时间窗的计算有太多的延迟。

由于这个原因，流媒体程序可能会显式地期望一些后期元素。延迟元素是在系统事件时间时钟(由水印表示)已经通过了延迟元素的时间戳之后到达的元素。有关如何在事件时间窗口中处理延迟元素的更多信息，请参见//todo。[https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/operators/windows.html\#allowed-lateness][https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_stream_operators_windows.html_allowed-lateness]

## Idling sources（闲置资源） ##

目前，使用纯事件时间水印生成器，如果没有要处理的元素，水印将无法进行处理。这意味着，在传入数据出现间隙的情况下，事件时间将不会进展，例如，窗口操作符将不会被触发，因此现有窗口将无法生成任何输出数据。

为了避免这种情况，可以使用不只是基于元素时间戳进行分配的周期性水印分配程序。一个示例解决方案可以是一个转让者，在一段时间没有观察到新事件之后，转换到使用当前处理时间作为时间基础。

可以使用SourceFunction.SourceContext\#markAsTemporarilyIdle将源标记为空闲。详情请参考该方法的Javadoc以及StreamStatus。

## 调试水印 ##

//todo

## 算子如何处理水印 ##

一般来说，操作人员需要完整地处理一个给定的水印，然后再将其转发到下游。例如，WindowOperator将首先评估应该触发哪个窗口，并且只有在生成由水印触发的所有输出之后，水印本身才会被发送到下游。换句话说，由于水印的出现而产生的所有元素将在水印之前发出。

同样的规则也适用于TwoInputStreamOperator。但是，在这种情况下，算子的当前水印被定义为其两个输入的最小值。

此行为的细节由OneInputStreamOperator\# process水印、两个oinputstreamoperator \#processWatermark1和两个oinputstreamoperator \#processWatermark2方法的实现定义。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70]: /images/20230601/3474f89601a245de8075ba5b22b636f0.png
[20191022162853607.png]: /images/20230601/a40cb50d42e04d22ac8a8e9432fe4f36.png
[20191022163037991.png]: /images/20230601/694cb8c5483742bfa8607b4b97c5ff39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 1]: /images/20230601/eb82a915c40e4575845572ca16857efa.png
[https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_stream_operators_windows.html_allowed-lateness]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/operators/windows.html#allowed-lateness