乐鱼电竞

教育行业A股IPO第一股（股票代码 003032）

全国咨询/投诉热线：400-618-4000

watermark的作用是什么?怎样保证数据不丢失?

更新时间:2023年10月10日10时52分 来源:乐鱼电竞 浏览次数:

好口碑IT培训

　　在大数据处理中，watermark是一种时间概念，用于衡量事件流数据的进度。它的作用是为了控制事件时间窗口的计算进度以及处理延迟。

　　具体而言，watermark可以把事件流数据按照事件发生的时间进度划分到不同的时间窗口中。在处理数据的过程中，必须要等到一个时间窗口的所有数据都到达后才能进行计算。而watermark就是用来判定一个时间窗口内的数据是否已经全量到达的标志。

　　保证数据不丢失的关键是通过合理设置watermark的生成和处理机制。在生成watermark的过程中，可以基于事件数据中的时间戳信息来确定watermark的位置。而在处理时，可以通过比较watermark和事件时间戳的关系，判断事件数据是否落后于watermark，如果落后则说明有数据丢失。

　　以下是使用Apache Flink的Java API示例代码，展示如何在流式处理中使用Watermark来控制事件时间窗口的计算进度。

// 导入必要的包
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

public class WatermarkExample {

    public static void main(String[] args) throws Exception {
        // 设置流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置时间特性为事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 创建数据源
        DataStream<Event> events = env.fromElements(
                new Event(1, "2021-01-01T00:00:00"),
                new Event(2, "2021-01-01T00:02:00"),
                new Event(3, "2021-01-01T00:01:30")
        );

        // 使用Watermark来指定事件时间
        events.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Event>() {
            private final long maxOutOfOrderness = 5000; // 最大乱序程度为5秒
            private long currentMaxTimestamp;

            @Override
            public long extractTimestamp(Event event, long previousElementTimestamp) {
                long timestamp = event.getTimestamp().toEpochMilli();
                currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                return timestamp;
            }

            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
            }
        });

        // 在这里添加更多的流处理操作，如窗口计算、聚合等

        // 执行流式处理
        env.execute("Watermark Example");
    }

    // 定义事件类
    public static class Event {
        private int id;
        private LocalDateTime timestamp;

        public Event(int id, String timestamp) {
            this.id = id;
            this.timestamp = LocalDateTime.parse(timestamp);
        }

        public int getId() {
            return id;
        }

        public LocalDateTime getTimestamp() {
            return timestamp;
        }
    }
}

　　在上面的示例中，我们首先设置了流式执行环境，并将时间特性设置为事件时间。然后，我们创建了一个包含三个事件的数据源，并为每个事件指定了事件时间戳。接下来，我们使用AssignerWithPeriodicWatermarks函数来为事件分配时间戳和Watermark。在这个函数中，我们定义了如何提取事件的时间戳，并根据最大乱序程度计算Watermark。最后，我们可以在assignTimestampsAndWatermarks方法后添加更多的流处理操作，如窗口计算、聚合等。

　　为了更好地保证数据不丢失，还可以采取一些策略来处理数据落后的情况，比如等待一段时间以等待可能的延迟数据到达，或者设置数据的最大乱序程度，超过乱序程度的数据将被丢弃。同时，还可以通过设置watermark的间隔时间来控制事件时间窗口的大�。允视Σ煌拇硌映傩枨�。

0 分享到：
和我们在线交谈！
【网站地图】【sitemap】