Flink中的创建Watermark水位线

在Flink中，水位线可大致分为乱序流水位线和有序流水位线。实际开发中用的最多的就是乱序流水位线
在此之前，你已了解Flink在分布式环境下Watermark的传播方式

Flink官方提供的设置水位线的方法有Source之前和Source之后，这里主要介绍Source之后的方法

默认方法

通过assignTimestampsAndWatermarks()方法来设置。
assignTimestampsAndWatermarks()方法主要依赖WatermarkStrategy，通过 WatermarkStrategy 我们可以为每条数据设置时间戳并生成 Watermark 。
assignTimestampsAndWatermarks() 方法源码

其中 WatermarkStrategy 实现了两个接口，分别是TimestampAssignerSupplier<T>设置时间戳和WatermarkGeneratorSupplier<T>生产水位线

基于使用场景的普遍，Flink官方已经封装好了两个方法供我们调用，大致使用方式如下：

dataStreamSource.assignTimestampsAndWatermarks(
                    WatermarkStrategy   
                      //.<T>forBoundedOutOfOrderness(...) // 乱序流    
                        .<T>forMonotonousTimestamps(...) // 有序流
                        .withTimestampAssigner(...)
                 );                

• forBoundedOutOfOrderness()/forMonotonousTimestamps(...)：为具体水位线生产的策略
• T：为数据流的元素类型
• withTimestampAssigner(...)：为指定事件时间戳

实际用法

package com.peng.time;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;


/**
 * @author 海绵先生
 * @Description TODO 生成简单的乱序流和有序流水位线
 * @date 2022/10/17-16:11
 */
public class WatermarkTest01 {
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Event {
        private String user;
        private String url;
        private Long timestamp;
    }
    public static void main(String[] args) {
        // TODO env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置时间水位线——生成间隔100毫秒
        env.getConfig().setAutoWatermarkInterval(100);

        // TODO source
        DataStreamSource<Event> dataStreamSource = env.fromElements(new Event("Sponge", "./user", 100L),
                new Event("Tom", "./user", 150L),
                new Event("Jrak", "./lookspouce", 300L),
                new Event("Sponge", "./like", 400L));

        // TODO 有序流的水平线生成方法
        //forMonotonousTimestamps是一个泛型方法，所以前面要指定一个泛型
        SingleOutputStreamOperator<Event> eventSingleOutputStreamOperator = dataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy
                //forMonotonousTimestamps 是WatermarkStrategy接口里的静态方法，为设置有序流的水位线
                .<Event>forMonotonousTimestamps()// 内部设置的是允许延迟时间为0，源码：super(Duration.ofMillis(0));返回的类型是WatermarkStrategy<T>
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    // 抽取时间戳的逻辑
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));
        /*
        * WatermarkStrategy.forMonotonousTimestamps()里只有一个水位线生成器，没有时间戳提取器
        * 所以还要在调用个withTimestampAssigner
        * */

        // TODO 乱序流生成水位线方法
        SingleOutputStreamOperator<Event> eventSingleOutputStreamOperator1 = dataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))//Duration.ofSeconds(2)允许迟到2秒
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));
    }
}

自定义水位线

通过实现WatermarkStrategy<T>接口下的WatermarkGenerator<T>、TimestampAssigner<T>两个方法

结合实例看看

package com.peng.time;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.common.metrics.stats.Max;

/**
 * @author 海绵先生
 * @Description TODO    自定义水位线
 * @date 2022/10/22-17:26
 */
public class WatermarkTest02 {
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Event {
        private String user;
        private String url;
        private Long timestamp;
    }

    public static void main(String[] args) throws Exception {
        // env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 更改默认水位线设置，设置成5000毫秒发射一次
        //env.getConfig().setAutoWatermarkInterval(5000L);

        // source
        DataStreamSource<Event> dataStreamSource = env.fromElements(new Event("Sponge", "./user", 100L),
                new Event("Tom", "./user", 150L),
                new Event("Jrak", "./lookspouce", 300L),
                new Event("Sponge", "./like", 400L));

        dataStreamSource.assignTimestampsAndWatermarks(new CustomWatermarkStrategy()).print();

        env.execute();

    }
    public static class CustomWatermarkStrategy implements WatermarkStrategy<Event> {

        @Override
        public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            // 该方法下要返回一个 WatermarkGenerator 的实例类
            return new CustomPeriodicGenerator();
        }

        @Override
        public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<Event>() {
                @Override
                public long extractTimestamp(Event element, long recordTimestamp) {
                    // 指定事件时间戳
                    return element.timestamp;
                }
            };
        }
    }

    public static class CustomPeriodicGenerator implements WatermarkGenerator<Event>{

        private Long delayTime = 5000L;//设置延迟时间 ms
        private Long maxTs = Long.MIN_VALUE + delayTime + 1L;//观察到的最大时间戳,先设个初始值

        //每来一条数据，就会调用一次onEvent()方法
        @Override
        public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {// output 为发射水位线对象
            maxTs = Math.max(event.timestamp,maxTs); //更新最大时间戳
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            // 发射水位线，默认200ms调用一次
            output.emitWatermark(new Watermark(maxTs - delayTime -1L));
        }
    }
}

/*
* 水位线是由上游广播到下游，下游的并行度会同时接收到上游水位线
*
* */

另外，通过上面的WatermarkGenerator实现类我们可以看到，生产水位线的方法有两种，一种为周期性的，一种为逐个式的。逐个式的生产为来一条就生产一个水位线，只需要将onPeriodicEmit方法里的 output.emitWatermark放到 onEvent 中就行了。