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 的新流,可以在其上执行基于事件时间的操作。

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部