一文读懂Flink的水印、水位线：动态处理乱序事件，实时数据处理的利器

后端

2023-09-04 18:42:45

Flink 的水印和水位线：实时数据处理的守护者

在当今数据驱动的世界中，实时处理不断涌入的数据至关重要。Apache Flink，一个领先的分布式流处理框架，提供了强大的工具来处理这些具有挑战性的任务。其中，Flink 的水印和水位线机制是实时数据处理的基石。

水印：跟踪事件时间的界标

想象一下一个快节奏的河流，其中事件如水滴般流过。Flink 的水印就像在这条河中插入的标志，指示着当前事件时间。它确保 Flink 不会处理过早的事件，从而避免了错误的计算。

水印可以由事件本身携带，也可以由 Flink 根据事件的时间戳动态生成。通过跟踪水印，Flink 可以确保处理中的事件都在指定的时间范围内。

水位线：确保可靠处理的界限

水位线是 Flink 根据水印计算出的一个移动界限。它指示哪些事件已经可以被可靠地处理，即使它们还没有全部到达。水位线通常落后于水印一段时间，为事件提供足够的时间到达 Flink 系统。

水印和水位线协同工作：无缝处理乱序事件

Flink 的水印和水位线密切协作，动态处理乱序事件。当事件到达时，Flink 会根据事件的时间戳生成水印。然后，系统会根据水印计算出水位线。

下游算子会根据水位线决定哪些事件可以被处理。这确保了 Flink 不会在水位线之前处理事件，防止了不准确的计算。

Flink 水印和水位线的好处：从容应对实时挑战

动态处理乱序事件： Flink 可以处理乱序到达的事件，即使事件顺序与发生顺序不同。
低延迟数据处理： 水印和水位线机制允许 Flink 根据水位线处理事件，而不需要等到所有事件都到达。
准确性： Flink 确保在水位线之前不会处理事件，保证了数据的准确性。

Flink 水印和水位线在实践中：示例时间

让我们通过一个示例来说明 Flink 水印和水位线的工作原理：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.twitter.TwitterSource;

public class FlinkWatermarksExample {

    public static void main(String[] args) throws Exception {

        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从 Twitter 流读取数据
        DataStream<TwitterElement> tweets = env.addSource(new TwitterSource());

        // 提取推文的时间戳作为事件时间
        DataStream<TwitterElement> tweetsWithTimestamps = tweets
            .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<TwitterElement>() {
                @Override
                public long extractAscendingTimestamp(TwitterElement element) {
                    return element.getCreatedAt();
                }
            });

        // 分组推文并计算每 10 秒内的推文数
        DataStream<Long> tweetCounts = tweetsWithTimestamps
            .keyBy(TwitterElement::getId)
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            .sum("tweetCount");

        // 打印结果
        tweetCounts.print();

        // 执行作业
        env.execute();
    }
}