Flink时间【事件时间、摄取时间、处理时间】与水印概述

偏执的太偏执、 2023-03-13 11:34 43阅读 0赞

## 一.事件时间 ##

事件时间是每个事件在其生产者上创建的时间。  
该时间通常在它们进入Flink之前嵌入到记录中，并且可以从每个记录中提取事件时间戳。在事件时间中，时间的进度取决于数据，而不取决于任何时钟。事件时间程序必须指定如何生成“ 事件时间水印”，这是一种表示事件时间进展的机制。

在理想情况下，事件时间处理将产生完全一致且确定的结果，而不管事件何时到达或它们的顺序如何。但是，除非已知事件是按时间戳（按时间戳）到达的，否则事件时间处理会在等待无序事件时产生一定的延迟。由于只能等待有限的时间，因此这限制了确定性事件时间应用程序的可用性。

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

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

## 二.摄取时间 ##

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

摄取时间从概念上讲介于事件时间和处理时间之间。与处理时间相比 ，获取稍微麻烦，但结果却更可预测。由于摄取时间使用稳定的时间戳（在源处分配了一次），因此对记录的不同窗口操作将引用相同的时间戳，而在处理时间中，每个窗口操作都可以将记录分配给不同的窗口（基于本地系统时钟和传输延迟）。

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

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

## 三.处理时间 ##

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

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

处理时间是最简单的时间概念，不需要流和机器之间的协调。它提供了最佳的性能和最低的延迟。但是，在分布式和异步环境中，处理时间不能提供确定性，因为它容易受到记录到达系统（例如从消息队列）到达系统的速度，记录在系统内部操作之间流动的速度的影响，以及中断（计划的或其他方式）等等。  
事件时间、摄取时间和处理时间图解如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]

## 四.设置时间特征 ##

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

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

## 五.水印 ##

支持事件时间的流处理器需要一种测量事件时间进度的方法。例如，当事件时间超过一个小时结束时，需要通知构建每小时窗口的窗口操作，以便该操作可以关闭正在进行的窗口。

事件时间可以独立于处理时间（由挂钟测量）进行。例如，在一个程序中，操作的当前事件时间可能会稍微落后于处理时间 （考虑到事件接收的延迟），而两者均以相同的速度进行。另一方面，另一个流媒体程序可以通过快速转发已经在Kafka主题（或另一个消息队列）中缓存的一些历史数据来在数周的事件时间内进行处理，而处理时间仅为几秒钟。

Flink中衡量事件时间进度的机制是水印。水印作为数据流的一部分流动，并带有时间戳t。水印（T）声明事件时间已在该流中达到时间t，这意味着该流中不应再有时间戳t<= t的元素（即，时间戳早于或等于水印的事件）。

下图显示了带有（逻辑）时间戳记的事件流，以及串联的水印。在此示例中，事件是按顺序排列的（相对于其时间戳），这意味着水印只是流中的周期性标记。  
![在这里插入图片描述][2020051310081991.png]  
水印对于乱序流至关重要，如下图所示，其中事件不是按其时间戳排序的。通常，水印是一种声明，即到流中的那个点，直到某个时间戳的所有事件都应该到达。一旦水印到达操作，操作就可以将其内部事件时钟提前到水印的值。  
![在这里插入图片描述][20200513100946355.png]  
请注意，事件时间是由新创建的一个（或多个）流元素从产生它们的事件或触发了创建这些元素的水印中继承的。

## 六.并行流中的水印 ##

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

随着水印在流式传输程序中的流动，它们会在事件时间之前到达其操作。每当操作提前其事件时间时，都会为其后续操作在下游生成新的水印。

一些运算符消耗多个输入流；例如，并集，或遵循keyBy（…）或partition（…）函数的运算符。该操作的当前事件时间是其输入流的事件时间中的最小值。随着其输入流更新其事件时间，操作也将更新。

下图显示了流过并行流的事件和水印的示例，操作跟踪事件时间。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]

## 七.后期元素 ##

某些元素可能会违反水印条件，这意味着即使在发生水印（t）之后，也会出现更多时间戳<= t的元素。实际上，在许多现实世界的设置中，某些元素可以任意延迟，从而无法指定某个事件时间戳标记的所有元素都将发生的时间。此外，即使可以限制延迟，通常也不希望将水印延迟太多，因为这会导致事件时间窗的评估延迟过多。

由于这个原因，流式传输程序可能会明确期望某些后期元素。延迟元素是在系统的事件时间时钟（由水印指示）已经经过延迟元素时间戳的时间之后到达的元素。

## 八.闲置来源 ##

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

为了避免这种情况，可以使用周期性的水印分配器，这些分配器不仅基于元素时间戳进行分配。一个示例解决方案可能是一个分配器，该分配器在一段时间内未观察到新事件之后切换为使用当前处理时间作为时间基础。

可以使用将源标记为空闲SourceFunction.SourceContext\#markAsTemporarilyIdle。

## 九.操作处理水印 ##

通常，要求操作在将给定水印转发到下游之前对其进行完全处理。例如， WindowOperator将首先评估应触发哪个窗口，只有在产生了所有由水印触发的输出之后，水印本身才会被发送到下游。换句话说，由于水印的出现而产生的所有元素将在水印之前发出。

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

这种行为的细节由的实现方式定义OneInputStreamOperator\#processWatermark， TwoInputStreamOperator\#processWatermark1和TwoInputStreamOperator\#processWatermark2方法。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: /images/20230312/d72124590d3e4213bf9983cb4647e486.png
[2020051310081991.png]: https://img-blog.csdnimg.cn/2020051310081991.png
[20200513100946355.png]: https://img-blog.csdnimg.cn/20200513100946355.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200513102450160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70