Flink Watermark详解

Watermark 是用于处理流数据中事件时间（event time）乱序情况的重要机制。在流处理中，数据往往不是按照它们实际发生的时间顺序到达的，这可能是由于网络延迟、系统处理延迟或其他因素导致的。为了能够在这种乱序环境中正确地执行基于时间的操作（如时间窗口聚合），Flink 引入了 Watermark 的概念。
Watermark 是一个特殊的标记，它表示“在此时间戳之前的数据应该都已经到达了”。当 Flink 的算子（operator）处理到 Watermark 时，它会认为该 Watermark 时间戳之前的所有数据都已经到达了，并可以安全地关闭或处理任何基于该时间戳的窗口。

概念

• **定义：**Watermark是一个特殊的时间戳，代表了某个时间点之前的数据理论上应该都已经到达了系统，即“最多允许的延迟”。
• **作用：**用于处理乱序事件，确保在某个时间窗口内完成所有相关的事件处理。

原理

• **乱序问题：**在流处理中，由于网络延迟等因素，事件可能会乱序到达。Watermark机制就是用来解决这种乱序问题。
• **工作原理：**当数据源在确认所有小于某个时间戳的消息都已输出到Flink流处理系统后，会生成一个包含该时间戳的Watermark，插入到消息流中。Flink operator算子按照时间窗口缓存所有流入的消息，当操作符处理到Watermark时，它会对所有小于该Watermark时间戳的时间窗口的数据进行处理并发送到下一个操作符节点，然后也将Watermark发送到下一个操作符节点。

用途

• **确保窗口计算的正确性：**Watermark结合窗口机制，可以确保在特定的时间后触发窗口去计算，从而避免因为乱序事件导致的窗口计算错误。
• **处理延迟数据：**Watermark提供了一个“最多允许的延迟”机制，对于延迟到达的数据，Flink可以根据Watermark来决定是否将其纳入当前窗口的计算。

样例

package com.wfg.flink.example.watermark;

import com.wfg.flink.example.watermark.data.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.time.Instant;

public class FlinkWatermarkDemo {

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 假设我们有一个数据源，这里使用 fromElements 模拟
        DataStream<Event> eventStream = env.fromElements(
                new Event(1L, Instant.now().minusSeconds(10).toEpochMilli()),
                new Event(2L, Instant.now().minusSeconds(8).toEpochMilli()),
                new Event(3L, Instant.now().minusSeconds(12).toEpochMilli()),
                new Event(4L, Instant.now().minusSeconds(15).toEpochMilli()),
                new Event(5L, Instant.now().minusSeconds(19).toEpochMilli()),
                new Event(6L, Instant.now().minusSeconds(18).toEpochMilli()),
                new Event(7L, Instant.now().minusSeconds(22).toEpochMilli())
        );

        // 定义 Watermark 策略，允许 5 秒的乱序
        WatermarkStrategy<Event> watermarkStrategy = WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> {
                    // 从事件中提取时间戳
                    return event.getTimestamp();
//                    timestamp.assignTimestamp(event.getTimestamp());
                });

        // 应用 Watermark 策略，并处理数据流
        DataStream<String> resultStream = eventStream
                .assignTimestampsAndWatermarks(watermarkStrategy)
                // 根据事件 ID 进行分区（这只是一个示例，实际可能根据业务需求分区）
                .keyBy(Event::getId)
                // 接下来可以进行窗口操作、时间聚合等操作
                .map(new MapFunction<Event, String>() {
                    @Override
                    public String map(Event event) throws Exception {
                        return "Event ID: " + event.getId() + ", Timestamp: " + Instant.ofEpochMilli(event.getTimestamp());
                    }
                });

        // 输出结果
        resultStream.print();

        // 执行任务
        env.execute("Flink Watermark Demo");
    }
}

若运行出错，可配置启动环境：–add-opens java.base/java.util=ALL-UNNAMED
数据类

package com.wfg.flink.example.watermark.data;

import lombok.Data;

/**
 * @author wfg
 */
@Data
public class Event {
    private final long id;
    private final long timestamp;

    public Event(long id, long timestamp) {
        this.id = id;
        this.timestamp = timestamp;
    }
}

WatermarkStrategy

atermarkStrategy是用于处理基于事件时间（event time）的流计算系统中可能出现的数据乱序情况的机制。
Watermark是数据流中的一种特殊数据，由Flink内部周期（可自定义）产生。它的主要作用是指示某个时间点之前的数据已经到达Flink系统，从而允许Flink开始处理这些数据。Watermark的生成策略可以实现数据乱序的兼容。

使用

atermarkStrategy在Flink中有两种主要的使用方式：

1. 直接在数据源上使用： 这种方式下，WatermarkStrategy会在数据源处被指定，并应用于从数据源读取的数据流。这种方式可以更精准地跟踪Watermark，因为数据源可以利用watermark生成逻辑中有关分片/分区的信息。
2. 直接在非数据源的操作之后使用： 如果无法直接在数据源上设置WatermarkStrategy，可以在数据流的其他位置（如经过某个操作后）设置。但这种方式通常不如第一种方式精准。

配置

WatermarkStrategy的配置主要涉及到Watermark的生成策略和自动发送周期。例如，可以使用WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20))来配置一个允许数据乱序程度不超过20秒的WatermarkStrategy。此外，还可以通过修改Flink的配置文件（如flink-conf.yaml）或调用相关API方法来设置Watermark的自动发送周期。

应用

基于Flink 1.16+版本的Java API，可以使用WatermarkStrategy类配合TimestampAssigner和TimestampExtractor接口来实现Watermark的生成器。具体实现方式可以参考相关文档和示例代码。

详情

WatermarkStrategy 是一个接口，它定义了如何为流中的事件生成 Watermarks。由于 Flink 是一个开源项目，我们可以直接查看其源代码来了解 WatermarkStrategy 的具体实现。
WatermarkStrategy 接口定义在 Flink 的 org.apache.flink.streaming.api.functions.timestamps 包中。这个接口定义了两个方法：

1. TimestampAssigner createTimestampAssigner(SerializedValue<TypeInformation> typeInfo): 用于创建一个 TimestampAssigner，该 TimestampAssigner 负责为流中的每个元素分配时间戳。
2. WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context): 用于创建一个 WatermarkGenerator，该 WatermarkGenerator 负责基于流中的元素生成 Watermarks。

// 定义 Watermark 策略，允许 5 秒的乱序
        WatermarkStrategy<Event> watermarkStrategy = WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> {
                    // 从事件中提取时间戳
                    return event.getTimestamp();
//                    timestamp.assignTimestamp(event.getTimestamp());
                });

通常，不会直接实现 WatermarkStrategy 接口，而是使用 Flink 提供的静态工厂方法来创建策略。例如，WatermarkStrategy.forBoundedOutOfOrderness(Duration maxOutOfOrderness) 方法就是一个常用的策略，它允许一定程度的乱序。

assignTimestampsAndWatermarks

assignTimestampsAndWatermarks 方法是用于为数据流中的事件分配时间戳和 Watermarks 的。这个方法通常与 WatermarkStrategy 一起使用，以定义如何为流中的每个元素分配时间戳以及何时生成 Watermarks。

// ...  
  
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
  
// 假设有一个名为 eventStream 的 DataStream，其中包含具有时间戳的事件  
DataStream<MyEvent> eventStream = ...; // 获取或创建事件流  
  
// 创建一个 WatermarkStrategy，这里使用了一个允许一定乱序的 BoundedOutOfOrdernessTimestampExtractor  
WatermarkStrategy<MyEvent> watermarkStrategy = WatermarkStrategy  
    .<MyEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))  
    .withTimestampAssigner(new BoundedOutOfOrdernessTimestampExtractor<MyEvent>() {  
        @Override  
        public long extractTimestamp(MyEvent element) {  
            return element.getTimestamp(); // 假设 MyEvent 有一个 getTimestamp() 方法返回事件的时间戳  
        }  
  
        @Override  
        public long getMaxAllowedLatency(MyEvent element) {  
            return Duration.ofSeconds(10).toMillis(); // 最大允许乱序时间为10秒  
        }  
    });  
  
// 为事件流分配时间戳和 Watermarks  
DataStream<MyEvent> timestampedStream = eventStream.assignTimestampsAndWatermarks(watermarkStrategy);  
  
// 现在可以基于事件时间进行窗口操作或其他时间感知的操作了  
timestampedStream  
    .keyBy(event -> event.getKey()) // 假设 MyEvent 有一个 getKey() 方法  
    .timeWindow(Time.seconds(30)) // 使用基于事件时间的30秒窗口  
    .apply(new WindowFunction<MyEvent, String, String, TimeWindow>() {  
        // ... 实现 WindowFunction  
    })  
    .print(); // 打印结果或其他后续操作  
  
// ...

WatermarkStrategy 使用了 BoundedOutOfOrdernessTimestampExtractor，它允许一定程度的数据乱序（在这个例子中是10秒）。extractTimestamp 方法用于为事件分配时间戳，而 getMaxAllowedLatency 方法定义了乱序时间的上限。然后，我们使用 assignTimestampsAndWatermarks 方法将这个策略应用到事件流上，从而得到一个带有时间戳和 Watermarks 的新流，可以在其上执行基于事件时间的操作。