Flink生成时间戳/水印

「爱情、让人受尽委屈。」 2023-02-17 03:47 19阅读 0赞

## 一.概述 ##

生成时间戳/水印与在事件时间有关。有关事件时间， 处理时间和摄取时间，参考：[事件时间概述][Link 1]

为了处理事件时间，流式传输程序需要相应地设置时间特征。

val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

## 二.分配时间戳 ##

为了使用事件时间，Flink需要知道事件的时间戳，这意味着流中的每个元素都需要分配其事件时间戳。这通常是通过从元素的某个字段访问/提取时间戳来完成的。

时间戳分配与生成水印齐头并进，水印告诉系统事件时间的进展。

有两种分配时间戳和生成水印的方法：

1.  直接在数据流源中生成。
2.  通过时间戳分配器/水印生成器：在Flink中，时间戳分配器还定义要发送的水印。

**注意**：自1970年1月1日T00：00：00Z的Java时代以来，时间戳和水印都指定为毫秒。

## 三.带有时间戳和水印的源函数 ##

流源可以将时间戳直接分配给它们产生的元素，并且它们还可以发出水印。完成此操作后，无需时间戳分配器。请注意，如果使用时间戳分配器，则源所提供的任何时间戳和水印都将被覆盖。

要将时间戳直接分配给源中的元素，源必须使用SourceContext上的collectWithTimestamp(…) 方法。要生成水印，源必须调用该emitWatermark(Watermark)函数。

下面是一个简单的示例（非检查点）源，该源分配时间戳并生成水印：

override def run(ctx: SourceContext[MyType]): Unit = { 
    	while (/* condition */) { 
    		val next: MyType = getNext()
    		ctx.collectWithTimestamp(next, next.eventTimestamp)
    
    		if (next.hasWatermarkTime) { 
    			ctx.emitWatermark(new Watermark(next.getWatermarkTime))
    		}
    	}
    }

## 四.时间戳分配器/水印生成器 ##

时间戳分配器获取流并产生带有时间戳的元素和水印的新流。如果原始流已经具有时间戳或水印，则时间戳分配器将覆盖它们。

时间戳分配器通常在数据源之后立即指定，但并非严格要求这样做。例如，一种常见的模式是在时间戳分配器之前解析（MapFunction）和筛选（FilterFunction）。无论如何，都需要在**事件时间的第一个操作**（例如第一个窗口操作）之前指定时间戳分配器。作为一种特殊情况，当使用Kafka作为流作业的源时，Flink允许在源（或使用者）内部指定时间戳分配器/水印发射器。

val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    val stream: DataStream[MyEvent] = env.readFile(
             myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
             FilePathFilter.createDefaultFilter())
    
    val withTimestampsAndWatermarks: DataStream[MyEvent] = stream
            .filter( _.severity == WARNING )
            .assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks())
    
    withTimestampsAndWatermarks
            .keyBy( _.getGroup )
            .timeWindow(Time.seconds(10))
            .reduce( (a, b) => a.add(b) )
            .addSink(...)

## 五.带有定期水印 ##

AssignerWithPeriodicWatermarks 分配时间戳并定期生成水印（可能取决于流元素，或仅基于处理时间）。

通过定义生成水印的间隔（每n毫秒） ExecutionConfig.setAutoWatermarkInterval(…)。分配器的getCurrentWatermark()方法每次都会被调用，如果返回的水印非空且大于前一个水印，则将发出新的水印。

在这里，我们显示了两个使用定期水印生成的时间戳分配器的简单示例。请注意，Flink附带与以下所示BoundedOutOfOrdernessTimestampExtractor类似的内容BoundedOutOfOrdernessGenerator。

/** * This generator generates watermarks assuming that elements arrive out of order, * but only to a certain degree. The latest elements for a certain timestamp t will arrive * at most n milliseconds after the earliest elements for timestamp t. */
    class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEvent] { 
    
        val maxOutOfOrderness = 3500L // 3.5 seconds
    
        var currentMaxTimestamp: Long = _
    
        override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = { 
            val timestamp = element.getCreationTime()
            currentMaxTimestamp = max(timestamp, currentMaxTimestamp)
            timestamp
        }
    
        override def getCurrentWatermark(): Watermark = { 
            // return the watermark as current highest timestamp minus the out-of-orderness bound
            new Watermark(currentMaxTimestamp - maxOutOfOrderness)
        }
    }
    
    /** * This generator generates watermarks that are lagging behind processing time by a fixed amount. * It assumes that elements arrive in Flink after a bounded delay. */
    class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent] { 
    
        val maxTimeLag = 5000L // 5 seconds
    
        override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = { 
            element.getCreationTime
        }
    
        override def getCurrentWatermark(): Watermark = { 
            // return the watermark as current time minus the maximum time lag
            new Watermark(System.currentTimeMillis() - maxTimeLag)
        }
    }

## 六.带标点的水印 ##

要在特定事件表明可能会生成新的水印时，请使用 AssignerWithPunctuatedWatermarks。对于此类，Flink将首先调用该extractTimestamp(…)方法为元素分配时间戳，然后立即checkAndGetNextWatermark(…)在该元素上调用该 方法。

该checkAndGetNextWatermark(…)方法会传递该方法中分配的时间戳extractTimestamp(…) ，并可以决定是否要生成水印。每当该checkAndGetNextWatermark(…) 方法返回一个非空水印，并且该水印大于最新的先前水印时，就会发出新的水印。

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[MyEvent] { 
    
    	override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = { 
    		element.getCreationTime
    	}
    
    	override def checkAndGetNextWatermark(lastElement: MyEvent, extractedTimestamp: Long): Watermark = { 
    		if (lastElement.hasWatermarkMarker()) new Watermark(extractedTimestamp) else null
    	}
    }

注意：可以在每个事件上生成水印。但是，由于每个水印都会在下游引起一些计算，因此过多的水印会降低性能。

## 七.Kafka分区的时间戳 ##

当使用Apache Kafka作为数据源时，每个Kafka分区可能都有一个简单的事件时间模式（升序时间戳或有界乱序）。但是，当使用来自Kafka的流时，通常会并行使用多个分区，从而交错插入分区中的事件并破坏每个分区的模式（这是Kafka的客户客户端工作方式所固有的）。

在这种情况下，可以使用Flink的Kafka分区感知水印生成。使用该功能，将在Kafka使用者内部针对每个Kafka分区生成水印，并且按与合并水印在流shuffle上相同的方式合并每个分区的水印。

例如，如果事件时间戳严格按照每个Kafka分区递增，则使用递增时间戳水印生成器生成按**分区的水印**将产生完美的整体水印。

下图显示了如何使用按kafka分区的水印生成，以及在这种情况下水印如何通过流数据流传播。

val kafkaSource = new FlinkKafkaConsumer09[MyType]("myTopic", schema, props)
    kafkaSource.assignTimestampsAndWatermarks(new AscendingTimestampExtractor[MyType] { 
        def extractAscendingTimestamp(element: MyType): Long = element.eventTimestamp
    })
    
    val stream: DataStream[MyType] = env.addSource(kafkaSource)

图解如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]

[Link 1]: https://blog.csdn.net/qq_27600723/article/details/106090412
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200612185145812.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70