深入Flink系列——watermark使用与源码详解_a watermark can not be defined for a processing-ti

前言

觉得文章有收获，欢迎关注公众号鼓励一下作者呀~
在学习的过程中，也搜集了一些量化、技术的视频及书籍资源，欢迎大家关注公众号【亚里随笔】获取

1 Flink时间体系

本节我们主要关注Flink的时间体系，包括Flink的时间语义、watermark机制及watermark的生成与传播原理，主要进行一些flink watermark理论知识的梳理。

1.1 Flink的时间语义

Flink支持三种时间概念：EventTime/ProcessingTime/IngestionTime，即事件时间、处理时间、摄入时间。

Flink的三种时间概念
EventTime是事件真实发生的时间。通常，事件时间就已经嵌入在记录中，在Flink系统中从记录中提取出事件时间。事件时间能够准确的反映事件发生的先后关系，它能够有效应对乱序事件、延迟事件。窗口的结果不会取决于数据流的读取或处理速度，而取决于数据。
ProcessingTime是执行相应算子操作的机器系统时间。在ProcessingTime下，所有基于时间的算子操（时间窗口）作将使用算子机器的系统时钟。通常，在窗口算子中使用处理时间会导致不确定的结果，因为窗口内容取决于元素到达的速率。同样的，在ProcessingTime的设置下，系统具有最低的延迟，因为此时处理任务无须依靠等待水位线来驱动事件时间前进。
IngestionTime指数据接入Flink系统的时间，将每个接收记录在数据源算子的处理时间作为事件时间的时间戳，是EventTime和ProcessingTime的混合体。但和EventTime相比，IngestionTime价值不大，因为它的性能和Event Time类似，但却无法提供确定的结果。只是当接入的事件不具体EventTime时可以借助IngestionTime来处理数据，自动分配时间戳和watermark。在实践中遇到一种比较特殊的情况，我认为应该也算作IngestionTime。当数据在进入消息队列时，消息队列的Connector会在Record上设置进入的时间戳，而Flink Source在基于Connector读取Record时，会读取该时间戳，用于设置Flink系统的时间戳和watermark。在这种情况下，虽然时间戳看似从数据中获得的，但本质上仍然是接入整个流处理系统时的时间，属于ingestionTime。
Flink时间概念对比

1.2 watermark机制

本质上，watermark提供了一个逻辑时钟，用来通知系统当前的事件时间。watermark用于在事件时间应用中推断每个任务当前的事件时间，基于时间的算子会使用这个时间来触发计算并推动进度前进。例如，基于时间窗口的任务会在其事件超过窗口结束边界时进行最终的窗口计算并发出结果。
watermark本质上一种时间戳，通常会基于watermark机制触发window窗口计算，用于处理乱序事件或延迟数据。watermark可以理解为全局进度指标，表示我们确信不会再有延迟事件到来的某个时间点。当一个算子接收到时间为T的水位线，就可以认为不会再接收到任何时间戳小于或等于T的事件了。而对于那些可能易于watermark的迟到事件，Flink中可以采取的机制有SideOutput、AllowedLateness或直接丢弃。

1.3 时间戳分配与watermark生成

时间戳和水位线通常都是在数据流刚刚进入流处理应用的时候分配和生成的。Flink DataStream可以通过三种方式完成时间戳分配与watermark生成工作。

• 在数据源完成。可以利用SourceFunction在应用读入数据流的时候分配时间戳和生成水位线。源函数发出一条记录，每个记录都可以附加一个时间戳，水位线可以作为特殊的记录在任何时间点发出。如果源函数（临时地）不再发出水位线，可以把自己声明为空闲。
• 周期性分配器（periodic assigner）。通过AssignerWithPeriodicWatermarks来从每条记录提取时间戳，并周期性的响应获取当前水位线的查询请求。提取出来的时间戳会附加到各自的记录上，查询到的水位线会注入到数据流中。
• 定点分配器（punctuated assigner）。通过AssignerWithPunctuatedWaters从数据流中根据某些特殊标记的记录来触发新的水位线，不会强制从每条记录中都提取一个时间戳。

1.4 watermark传播与事件时间

当任务接收到一个水位线时会执行以下操作：

1. 基于水位线记录的时间戳更新内部事件时间时钟。
2. 任务的时间服务会找出所有触发时间小于更新后事件时间的计时器。对于每个到期的计时器，调用回调函数。
3. 依赖根据更新后的事件时间将水位线发出。

Flink会将数据流划分为不同的分区，并由不同的算子任务来并行执行。每个分区作为一个数据流，都会包含带有时间戳的记录以及水位线。每个分区作为一条数据流，都会包含带有时间戳的记录以及水位线。根据算子上下流连接情况，其任务需要同时接收来自多个输入分区的记录和水位线，也可能将它们发送到多个输出分区。
如下图所示，一个任务会为它的每个输入分区都维护一个分区水位线（partition watermark）。当收到某个分区传来的水位线后，任务会以接收值和当前值中较大的那个去更新对应分区水位线的值。随后，任务会把事件时间时钟调整为所有分区水位线中的最小的那个值。如果事件时间时钟向前推动，任务会先处理因此而触发的计时器，之后才会把对应的水位线发往所有连接的输出分区，以实现事件时间到全部下游任务的广播。

Flink的水位线处理和传播算法保证了算子任务所发出的记录时间戳和水位线一定会对齐。然而，这依赖于一个事实：所有分区都会持续提供自增的水位线。只要有一个分区的水位线没有前进，或分区完全空闲下来不再发送任何记录或水位线，任务的事件时间时钟就不会前进，继而导致计时器无法触发。这种情况会给那些靠时钟前进来执行计算或清除状态的时间算子带来麻烦。如果一个任务没有从全部输入任务以常规间隔接收新的水位线，就会导致相关算子的处理延迟或状态大小激增。

2 watermark的使用示例

看Flinkv1.15已经在基于WatermarkStrategy/WatermarkGenerator接口来设置watermark了，为了整理学习的内容，目前还是先梳理旧的TimestampAssigner接口。
本节主要关注watermark如何在flink datastream api应用中使用，展示了基于AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks的示例。

2.1 时间特性设置

Flink支持三种时间属性TimeCharacteristic.EventTime/ProcessingTime/IngestionTime，可以通过StreamExecutionEnvironment#setStreamTimeCharacteristic方法来设置。
另外，Flink系统自动下发watermark的周期是可以设置的，通过ExecutionConfig#setAutoWatermarkInterval方法设置，默认autoWatermarkInterval=200L。
相关代码示例如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
// 设置系统主动轮询时间
ExecutionConfig config = env.getConfig();
config.setAutoWatermarkInterval(2000);
1
2
3
4
5
6
7

2.2 分配时间戳和watermark

Flink中可以通过三种方式抽取和生成Timestamp和watermark：

• 在SourceFunction中抽取和生成
• 在DataStream中作为独立算子抽取和生成
• 通过connector提供的接口抽取和生成

我们重点关注在DataStream数据转换过程中抽取Timestamp和生成watermark，由 DataStream API提供了两个接口来完成：AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks。AssignerWithPeriodicWatermarks的默认抽象实现类有AscendingTimestampExtractor和BoundedOutOfOrdernessTimestampExtractor。TimestampAssigner实现之间的UML关系图及其特性如下。

2.3 AssignerWithPeriodicWatermarks使用示例

这里直接使用AssignerWithPeriodicWatermarks，参考网上的博客，整理了一个使用示例。AscendingTimestampExtractor和BoundedOutOfOrdernessTimestampExtractor的使用更简单，只需要实现extractTimestamp接口即可。
示例说明：

• 使用env.fromElements从String[] 中模拟流式数据
• 设置watermark，实现AssignerWithPeriodicWatermarks
• 进行map转换，将String转换为Tuple2<String, String>，根据key分组
• 使用TumblingEventTimeWindows，将5秒内的同组数据，聚合后输出
• 注意：flink系统周期性的生成watermark时，才会定时调用getCurrentWatermark，在没有调用之前，watermark是不会变的，在示例中也进行了一些主动调用和周期性生成watermark的测试。

public class WatermarkEventData {
    public static String[] eventDataInOder =
            new String[] {
                "HelloWaterMark,1553503185000",
                "HelloWaterMark,1553503186000",
                "HelloWaterMark,1553503187000",
                "HelloWaterMark,1553503188000",
                "HelloWaterMark,1553503189000",
                "HelloWaterMark,1553503190000",
            };

    public static String[] eventDataOutOfOrder =
            new String[] {
                "HelloWaterMark,1553503185000",
                "HelloWaterMark,1553503186000",
                "HelloWaterMark,1553503187000",
                "HelloWaterMark,1553503188000",
                "HelloWaterMark,1553503189000",
                "HelloWaterMark,1553503190000",
                "HelloWaterMark,1553503187000",
                "HelloWaterMark,1553503186000",
                "HelloWaterMark,1553503191000",
                "HelloWaterMark,1553503192000",
                "HelloWaterMark,1553503193000",
                "HelloWaterMark,1553503194000",
                "HelloWaterMark,1553503195000",
            };
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public class OutOfOrderForPeriodicWatermark {
    public static void main(String[] args) throws Exception {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        // env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
        // 设置系统主动轮询时间
        ExecutionConfig config = env.getConfig();
        config.setAutoWatermarkInterval(2000);

        boolean isActiveCall = false;

        DataStream<String> dataStream =
                env.fromElements(WatermarkEventData.eventDataOutOfOrder)
                        .assignTimestampsAndWatermarks(
                                new AssignerWithPeriodicWatermarks<String>() {
                                    long currentTimestamp = 0L;
                                    long maxDelayAllowed = 0L;
                                    long currentWatermark;

                                    @Nullable
                                    @Override
                                    public Watermark getCurrentWatermark() {
                                        currentWatermark = currentTimestamp - maxDelayAllowed;
                                        return new Watermark(currentWatermark);
                                    }

                                    @Override
                                    public long extractTimestamp(
                                            String element, long recordTimestamp) {
                                        String[] arr = element.split(",");
                                        long timestamp = Long.parseLong(arr[1]);
                                        currentTimestamp = Math.max(timestamp, currentTimestamp);

                                        // 通过getCurrentWatermark实时获取watermark，而不是基于系统时间服务周期性调用
                                        if (!isActiveCall) {
                                            System.out.println(
                                                    "Key:"
                                                            + arr[0]
                                                            + ",EventTime: "
                                                            + simpleDateFormat.format(timestamp)
                                                            + ",上一条数据的水位线(系统轮询): "
                                                            + simpleDateFormat.format(
                                                                    currentWatermark));
                                        } else {
                                            System.out.println(
                                                    "Key:"
                                                            + arr[0]
                                                            + ",EventTime: "
                                                            + simpleDateFormat.format(timestamp)
                                                            + ",上一条数据的水位线(主动获取): "
                                                            + simpleDateFormat.format(
                                                                    Objects.requireNonNull(
                                                                                    getCurrentWatermark())
                                                                            .getTimestamp()));
                                        }

                                        try {
                                            Thread.sleep(1000);
                                        } catch (InterruptedException e) {
                                            e.printStackTrace();
                                        }
                                        return timestamp;
                                    }
                                });

        dataStream
                .map(
                        new MapFunction<String, Tuple2<String, String>>() {
                            @Override
                            public Tuple2<String, String> map(String value) throws Exception {
                                return new Tuple2<>(value.split(",")[0], value.split(",")[1]);
                            }
                        })
                .keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(
                        new AggregateFunction<Tuple2<String, String>, String, String>() {
                            @Override
                            public String createAccumulator() {
                                return "Start: ";
                            }

                            @Override
                            public String add(Tuple2<String, String> value, String accumulator) {
                                return accumulator
                                        + "-"
                                        + simpleDateFormat.format(Long.parseLong(value.f1));
                            }

                            @Override
                            public String getResult(String accumulator) {
                                return accumulator;
                            }

                            @Override
                            public String merge(String a, String b) {
                                return a + "-" + b;
                            }
                        })
                .print();

        env.execute("watermark test demo");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

Key:HelloWaterMark,EventTime: 2019-03-25 16:39:45,上一条数据的水位线(主动获取): 2019-03-25 16:39:45
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:46,上一条数据的水位线(主动获取): 2019-03-25 16:39:46
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:47,上一条数据的水位线(主动获取): 2019-03-25 16:39:47
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:48,上一条数据的水位线(主动获取): 2019-03-25 16:39:48
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:49,上一条数据的水位线(主动获取): 2019-03-25 16:39:49
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:50,上一条数据的水位线(主动获取): 2019-03-25 16:39:50
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:47,上一条数据的水位线(主动获取): 2019-03-25 16:39:50
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:46,上一条数据的水位线(主动获取): 2019-03-25 16:39:50
3> Start: -2019-03-25 16:39:45-2019-03-25 16:39:46-2019-03-25 16:39:47-2019-03-25 16:39:48-2019-03-25 16:39:49-2019-03-25 16:39:47
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:51,上一条数据的水位线(主动获取): 2019-03-25 16:39:51
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:52,上一条数据的水位线(主动获取): 2019-03-25 16:39:52
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:53,上一条数据的水位线(主动获取): 2019-03-25 16:39:53
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:54,上一条数据的水位线(主动获取): 2019-03-25 16:39:54
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:55,上一条数据的水位线(主动获取): 2019-03-25 16:39:55
3> Start: -2019-03-25 16:39:50-2019-03-25 16:39:51-2019-03-25 16:39:52-2019-03-25 16:39:53-2019-03-25 16:39:54
3> Start: -2019-03-25 16:39:55
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

Key:HelloWaterMark,EventTime: 2019-03-25 16:39:45,上一条数据的水位线(系统轮询): 1970-01-01 08:00:00
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:46,上一条数据的水位线(系统轮询): 1970-01-01 08:00:00
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:47,上一条数据的水位线(系统轮询): 1970-01-01 08:00:00
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:48,上一条数据的水位线(系统轮询): 1970-01-01 08:00:00
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:49,上一条数据的水位线(系统轮询): 1970-01-01 08:00:00
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:50,上一条数据的水位线(系统轮询): 1970-01-01 08:00:00
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:47,上一条数据的水位线(系统轮询): 2019-03-25 16:39:50
3> Start: -2019-03-25 16:39:45-2019-03-25 16:39:46-2019-03-25 16:39:47-2019-03-25 16:39:48-2019-03-25 16:39:49
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:46,上一条数据的水位线(系统轮询): 2019-03-25 16:39:50
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:51,上一条数据的水位线(系统轮询): 2019-03-25 16:39:50
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:52,上一条数据的水位线(系统轮询): 2019-03-25 16:39:51
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:53,上一条数据的水位线(系统轮询): 2019-03-25 16:39:51
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:54,上一条数据的水位线(系统轮询): 2019-03-25 16:39:51
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:55,上一条数据的水位线(系统轮询): 2019-03-25 16:39:54
3> Start: -2019-03-25 16:39:50-2019-03-25 16:39:51-2019-03-25 16:39:52-2019-03-25 16:39:53-2019-03-25 16:39:54
3> Start: -2019-03-25 16:39:55
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2.4 AssignerWithPunctuatedWatermarks使用示例

参考AssignerWithPeriodicWatermark的示例，应用AssignerWithPunctuatedWatermarks，实现AssignerWithPunctuatedWatermarks#checkAndGetNextWatermark和TimestampAssigner#extractTimestamp接口。Flink系统在运行的时候，会先调用extractTimestamp实现，提取数据中的timestamp；紧接着会调用checkAndGetNextWatermark实现，根据数据中的特殊标记生成watermark并下发；后续流程中，系统会保证比之前watermark大的watermark才会下发到下游节点。
在示例中，会从数据中的第3个字段是否是偶数来判断是否要生成watermark。

public class PunctuatedWatermark {
    public static void main(String[] args) throws Exception {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        // env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
        // 设置系统主动轮询时间
        ExecutionConfig config = env.getConfig();
        config.setAutoWatermarkInterval(2000);

        DataStream<String> dataStream =
                env.fromElements(WatermarkEventData.eventDataOutOfOrder)
                        .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<String>() {
                            @Nullable
                            @Override
                            public Watermark checkAndGetNextWatermark(String s, long l) {
                                String[] arr = s.split(",");
                                int flagInt = Integer.parseInt(arr[2]);

                                boolean ommitWatermark = flagInt % 2 == 0;
                                if (ommitWatermark){
                                    System.out.println(
                                            "Key:"
                                                    + arr[0]
                                                    + ",EventTime: "
                                                    + simpleDateFormat.format(l)
                                                    + ",水位线标识: "
                                                    + flagInt
                                                    + ",watermark: "
                                                    + simpleDateFormat.format(l));
                                    return new Watermark(l);
                                } else {
                                    System.out.println(
                                            "Key:"
                                                    + arr[0]
                                                    + ",EventTime: "
                                                    + simpleDateFormat.format(l) + "水位线标识: " + flagInt);
                                    return null;
                                }
                            }

                            @Override
                            public long extractTimestamp(String s, long l) {
                                String[] arr = s.split(",");
                                long timestamp = Long.parseLong(arr[1]);
                                try {
                                    Thread.sleep(1000);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                return timestamp;
                            }
                        });

        dataStream
                .map(
                        new MapFunction<String, Tuple2<String, String>>() {
                            @Override
                            public Tuple2<String, String> map(String value) throws Exception {
                                return new Tuple2<>(value.split(",")[0], value.split(",")[1]);
                            }
                        })
                .keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(
                        new AggregateFunction<Tuple2<String, String>, String, String>() {
                            @Override
                            public String createAccumulator() {
                                return "Start: ";
                            }

                            @Override
                            public String add(Tuple2<String, String> value, String accumulator) {
                                return accumulator
                                        + "-"
                                        + simpleDateFormat.format(Long.parseLong(value.f1));
                            }

                            @Override
                            public String getResult(String accumulator) {
                                return accumulator;
                            }

                            @Override
                            public String merge(String a, String b) {
                                return a + "-" + b;
                            }
                        })
                .print();

        env.execute("watermark test demo——punctuatedWatermark.");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

Key:HelloWaterMark,EventTime: 2019-03-25 16:39:45水位线标识: 1
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:46,水位线标识: 2,watermark: 2019-03-25 16:39:46
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:47水位线标识: 3
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:48,水位线标识: 4,watermark: 2019-03-25 16:39:48
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:49水位线标识: 5
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:50,水位线标识: 6,watermark: 2019-03-25 16:39:50
3> Start: -2019-03-25 16:39:45-2019-03-25 16:39:46-2019-03-25 16:39:47-2019-03-25 16:39:48-2019-03-25 16:39:49
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:47水位线标识: 7
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:46,水位线标识: 8,watermark: 2019-03-25 16:39:46
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:51水位线标识: 9
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:52,水位线标识: 10,watermark: 2019-03-25 16:39:52
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:53水位线标识: 11
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:54,水位线标识: 12,watermark: 2019-03-25 16:39:54
Key:HelloWaterMark,EventTime: 2019-03-25 16:39:55水位线标识: 13
3> Start: -2019-03-25 16:39:50-2019-03-25 16:39:51-2019-03-25 16:39:52-2019-03-25 16:39:53-2019-03-25 16:39:54
3> Start: -2019-03-25 16:39:55
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

3 watermark源码梳理

本节主要进行flink框架关于watermark实现源码的梳理，先对watermark数据结构进行介绍，然后简要介绍一下flink运行时各执行模块是如何调用的，最后梳理三种watermark生成方式中flink系统的处理流程。

3.1 watermark数据结构

watermark的功能是告诉flink系统：不会再有小于或等于watermark.timestamp的数据到达了。watermark本质上还是一个时间戳。从Flink的Watermark数据结构来看，唯一有意义的成员变量就是timestamp。

public final class Watermark extends StreamElement {
    public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);
    public static final Watermark UNINITIALIZED = new Watermark(Long.MIN_VALUE);
    /** The timestamp of the watermark in milliseconds. */
    private final long timestamp;
    public Watermark(long timestamp) {
        this.timestamp = timestamp;
    }
    public long getTimestamp() {
        return timestamp;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

3.2 flink运行模型简介

在梳理SouceFunction、DataStramp算子、connector提取watermark的处理流程前，我们先简单介绍下flink的运行模型，即用户写的UserFunction是如何被Flink加载然后在Runtime中运行的。
Flink DataStream构造的过程中，不同类型的转换操作都是按同样的方式进行的：首先将用户自定义的函数封装到Operator中，然后将Operator封装到Transformation结构中，最后将Transformation写入StreamExecutionEnvironment提供的Transformation List中。通过DataStream之间的转换操作构造StreamGraph数据结构，最终通过StreamGraph生成JobGraph并提交到集群上运行。在集群上运行时，首先在JobMaster中将JobGraph结构转换为ExecutionGraph，并且对ExecutionGraph中的Execution Vertiex节点进行调度和执行，最后将ExecutionVertex以Task的形式在TaskExecutor上运行。
Flink DataStream中，用户自定义UDF最终被调用的流程大致如下图所求。

3.3 SourceFunction提取watermark流程

在SourceFuntion中读取数据元素时，SourceContext接口中定义了抽取Timestamp和生成watermark的方法，如collectWithTimestamp和emitWatermark。当Flink作业基于EventTime时，就会使用StreamSourceContext.ManualWatermarkContext处理Watermark信息。
WatermarkContext.collectWithTimestamp方法由从Source算子接入的数据中抽取事件时间戳信息来设置元素的timestamp。生成watermark主要是通过调用WatermarkContext.emitWatermark()方法进行的。生成的Watermark首先会更新当前Source处子中的CurrentWatermark，然后将Watermark传递给下游算子继续处理。当下游算子接收到Watermark事件后，也会更新当前算子内部的CurrentWatermark。在WatermarkContext.emitWatermark()方法中会调用processAndEmitWatermark()方法将生成的watermark实时发送到下游算子中继续处理。不同的WatermarkContext子类，对processAndEmitWatermark的实现不同。
我们借助flink源码中SideOutputITCase里的自定义DataSource来梳理一下SourceFunction内的底层时间处理逻辑，SourceFunction的使用如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Integer> dataStream =
    env.addSource(
    new SourceFunction<Integer>() {
        private static final long serialVersionUID = 1L;

        @Override
        public void run(SourceContext<Integer> ctx) throws Exception {
            ctx.collectWithTimestamp(1, 0);
            ctx.emitWatermark(new Watermark(0));
            ctx.collectWithTimestamp(2, 1);
            ctx.collectWithTimestamp(5, 2);
            ctx.emitWatermark(new Watermark(2));
            ctx.collectWithTimestamp(3, 3);
            ctx.collectWithTimestamp(4, 4);
        }

        @Override
        public void cancel() {}
    });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

可以看出，在自定义SourceFunction时， 需要实现run和cancal方法，run方法可以获取到SourceContext，通过SourceContext的collect方法可以下发无timestamp的数据；通过collectWithTimeStamp方法，可以下发带timestamp的数据；通过emitWatermark方法可以下发Watermark。
SourceFuntion接口的定义如下：

public interface SourceFunction<T> extends Function, Serializable {
    
    void run(SourceContext<T> ctx) throws Exception;
    
    void cancel();

    interface SourceContext<T> {

        void collect(T element);

        @PublicEvolving
        void collectWithTimestamp(T element, long timestamp);

        @PublicEvolving
        void emitWatermark(Watermark mark);

        @PublicEvolving
        void markAsTemporarilyIdle();

        Object getCheckpointLock();

        void close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

接下来我们梳理一下env.addSource(new SouceFunction…)的源码间调用关系，如下图所示。详细源码就不贴了。

• 自定义实现SourceFunction接口，实现run方法。run方法内可以基于SourceContext向下游发送纯数据(collect)、发送带时间戳的数据(collectWithTimestamp)、发送watermark(emitWatermark)。SouceContext有ManualWatermarkContext和AutomaticWatermarkContext两种实现，根据TimeCharacteristic创建相应的SourceContext。
• 通过env.addSource方法基于自定义SourceFunction创建StreamSource，StreamSource是StreamOperator的子类，用于运行时供StreamTask调用。StreamSource的run方法会根据TimeCharacteristic创建SourceContext，调用userFunction.run(ctx)。
• StreamSource在运行时由SourceStreamTask调用。SourceStreamTask的processInput方法调用source.Thread()开启LegacySourceFunction的run方法。LegacySourceFunction的run方法调用StreamSource的run方法，进而调用自定义的userFunction.run方法。

3.4 DataStream算子提取watermark流程

org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator
除了能够在SourceFunction中直接分配Timestamp和生成Watermak，也可以在DataStream数据转换过程中进行相应操作，此时转换操作对应的算子就能使用生成的Timestamp和Watermark信息了。在DataStream算子中提取watermark的示例和方法特性在第2节中已经详细介绍了。这里我们就以AssignerWithPeriodicWatermarks和AssignerWithPuncatedWatermarks为例来梳理DataStream算子提取watermark的源码间调用关系，如下图所示。

• 自定义AssignerWithPeriodicWatermarks和AssignerWithPuncatedWatermarks
  • AssignerWithPeriodicWatermarks接口需要用户实现getCurrentWatermark和extractTimestamp函数，extractTimestamp函数当每处理一条记录时都会被调用，getCurrentWatermark被封装进AssignerWithPeridocWatermarkAdapter的onPeridoc Emit函数，当按watermarkInterval设定的processing Timer到期以后周期性的调用。
  • AssignerWithPunctuatedWatermarks接口需要用户实现checkAndGetNextWatermark和extractTimestamp函数，extractTimestamp函数来自于TimestampeAssigner，也是在处理每一条记录时都会被调用。checkAndGetNextWatermark函数被封装进AssignerWithPunctuatedWatermarksAdapter的onEvent函数，在extractorTimestamp之后被调用，根据需要向下游发送watermark。
• AssignerWithPeriodicWatermarks和AssignerWithPuncatedWatermarks通过datastream.assignTimestampAndWatermarks函数，被封装进TimestampAndWatermarkTransformation。由于在flink 1.15中AssignerWithPeriodicWatermarks和AssignerWithPuncatedWatermarks接口已经被升级为watermarkGenerator，flink 1.15中也提供了对应的Adapter将AssignerWithPeriodicWatermarks封装成AssignerWithPeriodicWatermarksAdapter，将AssignerWithPuncatedWatermarks封装成AssignerWithPuncatedWatermarksAdapter。
  • AssignerWithPeriodicWatermarksAdapter中实现了onPeriodicEmit函数，它直接调用AssignerWithPeriodicWatermarks.getCurrentWatermark，计算产生watarmark。onPeriodicEmit函数由flink系统按processingTimer周期性调用。AssignerWithPeriodicWatermarksAdapter的onEvent函数方法体为空，不做任何处理。
  • AssignerWithPuncatedWatermarksAdapter中实现了onEvent函数，它直接调用AssignerWithPunctuatedWatermarks.checkAndGetNextWatermark，根据需要计算产生watermark。onEvent函数在每条记录处理时都会被调用。
• TimestampAndWatermarkTransformation被TimestampsAndWatermarksTranformationTranslator转换成对应的TimestampAndWatermarkOperator，由运行时被OneInputStreamTask调用。
  • TimestampAndWatermarkOperator的processElement方法会调用timestampAssigner.extractTimestamp和watermarkGenerator.onEvent函数，根据Assigner类型调用至用户自定义的checkAndGetNextWatermark函数。processElement方法由OneInputStreamTask的emitRecord函数运行时调用。
  • TimestampAndWatermarkOperator的onProcessingTime方法会调用watermarkGenerator.onPeriodicEmit函数，相应地调用至自定义的getCurrentWatermark。在调用完watermarkGenerator.onPeriodicEmit函数之后，会调用registerTimer，根据watermarkInterval基于ProcessingTime设置定时器，周期性地回调。onProcessingTime方法由StreamTask的invokeProcessingTimeCallback进行回调待确定。
• TimestampAndWatermarkOperator被封装进OneInputStreamTask，运行时直接调用。
  • TimestampAndWatermarkOperator的emitRecord方法会调用TimestampAndWatermarkOperator的processElement函数。
  • TimestampAndWatermarkOperator的invokeProcessingTimeCallback方法，会调用TimestampAndWatermarkOperator的onProcessingTime函数[待确定]。

这里也展示一下TimestampAndWatermarkOperator里的关键调用代码。

public class TimestampsAndWatermarksOperator<T>... {
        ...
    @Override
    public void processElement(final StreamRecord<T> element) throws Exception {
        final T event = element.getValue();
        final long previousTimestamp =
                element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE;
        // timestampAssigner对应AssignerWithPeriodicWatermarks和AssignerWithPuncatedWatermarks
        final long newTimestamp = timestampAssigner.extractTimestamp(event, previousTimestamp);

        element.setTimestamp(newTimestamp);
        output.collect(element);
        // watermarkGenerator对应AssignerWithPeriodicWatermarksAdapter和AssignerWithPuncatedWatermarksAdapter
        watermarkGenerator.onEvent(event, newTimestamp, wmOutput);
    }

    @Override
    public void onProcessingTime(long timestamp) throws Exception {
        watermarkGenerator.onPeriodicEmit(wmOutput);
        
        final long now = getProcessingTimeService().getCurrentProcessingTime();
        getProcessingTimeService().registerTimer(now + watermarkInterval, this);
    }
        ...
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

3.5 通过connector的接口提取watermark流程

对于某些内置的数据源连接器来讲，是通过实现SourceFunction接口接入外部数据的，此时用户无法直接获取SourceFuntion的接口方法，会造成无法在SourceOperator中直接生成EventTime和Watermark的情况。但在一些数据源连接器中，如FlinkKafakaConsumer中，已经支持用户将AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks实现类传递到连接器的接口中，然后再通过连接器应用在对应的SourceFunction中，进而生成EventTime和Watermark。FlinkKafakaConsumer接口使用示例如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

FlinkKafkaConsumer<Long> kafkaSource =new FlinkKafkaConsumer<>(
    topic, new KafkaITCase.LimitedLongDeserializer(), standardProps);
kafkaSource.assignTimestampsAndWatermarks(
    new AssignerWithPunctuatedWatermarks<Long>() {
        private static final long serialVersionUID = -4834111173247835189L;
        @Nullable
        @Override
        public Watermark checkAndGetNextWatermark(
            Long lastElement, long extractedTimestamp) {
            if (lastElement % 11 == 0) {
                return new Watermark(lastElement);
            }
            return null;
        }
        
        @Override
        public long extractTimestamp(Long element, long previousElementTimestamp) {
            return previousElementTimestamp;
        }
    });

DataStream<Long> stream = env.addSource(kafkaSource);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

FlinkKafakaConsumer通过assignTimestampsAndWatermarks方法将AssignerWithPunctuatedWatermarks和AssignerWithPeriodicWatermarks实现类传入SourceFuntion中。同样，我们梳理下源码间的调用关系图，从kafkaSource.assignTimestampAndWatermarks开始至调用到extractTimestamp、onEvent和onPeriodicEmit，和以前面重复的地方就不展开画了。

• kafkaSource.assignTimestampsAndWatermarks中将AssignerWithPunctuatedWatermarks或AssignerWithPeriodicWatermarks经对应的Adapter生成WatermarkStrategy，这一步和datastream.assignTimestampsAndWatermarks是一致的。
• 在执行FlinkKafkaConsumerBase.run时，会调用createrFetcher函数将WatermarkStrategy构造KafkaFetcher。在构造的过程中调用createPartitionStateHolder，基于WatermarkStrategy构造KafkaTopicpartitionStateWithWatermarkGenerator，后续会基于partitionState进行时间戳与watermark的相关操作。
• 随后，FlinkKafkaConsumerBase.run会执行KafkaFetch.runFetchLoop，开始Fetch数据，其中会执行kafkaFetcher.partitionConsumerRecordsHandler和kafkaFetcher.emitRecordsWithTimestamps，最终会调用到partitionState.extractTimestamp和partitionState.onEvent。partionState就是在创建KafkaFetcher时由watermarStrategy构建而来的KafkaTopicPartitionStateWithWatermarkGenerator。同样，onEvent函数在每处理一条记录时都会被调用。
• 另外，AbstractFetcher类有一个内部类PeriodicWatermarkEmiter，PeriodicWatermarkEmiter的onProcessingTime函数是基于ProcessingTimer的回调函数，其中执行partitionState.onPeriodicEmit。并且对于KafkaConsumer而言，由于其存在多个partition，需要避免多分区而产生watermark不一致的情况。所以在周期性调用onProcessingTime下发watermark时，会计算所有分区watermark的最小值并下发。多分区watermark的计算和下发逻辑由watermarkOutputMultiplexer.onPeriodicEmit实现。onProcessintTime回调函数由timer周期性调用，先会在每个partition中调用KafkaTopicpartitionStateWithWatermarkGenerator.onPeriodicEmit，即用户自定义的AssignerWithPeriodicWatermarks接口，然后调用watermarkOutputMultiplexer.onPeriodicEmit。在这种情况下，即使用户没有实现AssignerWithPeriodicWatermarks接口，只是onPeriodicEmit为空，但不影响多分区watermark的下发处理逻辑。

这里也展示一下AbstractFetcther中的关键代码。

    protected void emitRecordsWithTimestamps(
            Queue<T> records,
            KafkaTopicPartitionState<T, KPH> partitionState,
            long offset,
            long kafkaEventTimestamp) {
        // emit the records, using the checkpoint lock to guarantee
        // atomicity of record emission and offset state update
        synchronized (checkpointLock) {
            T record;
            while ((record = records.poll()) != null) {
                long timestamp = partitionState.extractTimestamp(record, kafkaEventTimestamp);
                sourceContext.collectWithTimestamp(record, timestamp);

                // this might emit a watermark, so do it after emitting the record
                partitionState.onEvent(record, timestamp);
            }
            partitionState.setOffset(offset);
        }
    }
   private static class PeriodicWatermarkEmitter<T, KPH> implements ProcessingTimeCallback {
        @Override
        public void onProcessingTime(long timestamp) {

            synchronized (checkpointLock) {
                for (KafkaTopicPartitionState<?, ?> state : allPartitions) {
                    state.onPeriodicEmit();
                }
                // 多分区watermark处理逻辑
                watermarkOutputMultiplexer.onPeriodicEmit();
            }

            // schedule the next watermark
            timerService.registerTimer(timerService.getCurrentProcessingTime() + interval, this);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

3.6 flink算子间watermark传播源码梳理

前面梳理完各种timestamp提取和watermark设置的相关源码之后，我们现在梳理一下算子间watermark在传播时所经过的处理，也就是算子A向算子B传播过程中watermark对齐所经历的min-max操作。
在考虑partition的情况下，算子A向算子B的channel发送一条watermark，org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput#processElement方法会根据SteamElement的类型，执行statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), flattenedChannelIndices.get(lastChannel), output);
算子B的多partition watermark对齐逻辑就在inputWatermark中，代码如下。

public void inputWatermark(Watermark watermark, int channelIndex, DataOutput<?> output)throws Exception {
    // ignore the input watermark if its input channel, or all input channels are idle (i.e.
    // overall the valve is idle).
    if (lastOutputWatermarkStatus.isActive()
        && channelStatuses[channelIndex].watermarkStatus.isActive()) {
        long watermarkMillis = watermark.getTimestamp();

        // if the input watermark's value is less than the last received watermark for its input
        // channel, ignore it also.
        if (watermarkMillis > channelStatuses[channelIndex].watermark) {
            channelStatuses[channelIndex].watermark = watermarkMillis;

            // previously unaligned input channels are now aligned if its watermark has caught
            // up
            if (!channelStatuses[channelIndex].isWatermarkAligned
                && watermarkMillis >= lastOutputWatermark) {
                channelStatuses[channelIndex].isWatermarkAligned = true;
            }

            // now, attempt to find a new min watermark across all aligned channels
            findAndOutputNewMinWatermarkAcrossAlignedChannels(output);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

算子B的当前分区收到watermark以后，如果到达的watermark比当前分区的watermark的大，则更新当前分区的watermark。然后由findAndOutputNewMinWatermarkAcrossAlignedChannels函数遍历所有的分区，取各分区watermark的最小值来对齐各分区的watermark，如果对齐后的watermark往前推进了则下发，代码如下。

private void findAndOutputNewMinWatermarkAcrossAlignedChannels(DataOutput<?> output)
    throws Exception {
    long newMinWatermark = Long.MAX_VALUE;
    boolean hasAlignedChannels = false;

    // determine new overall watermark by considering only watermark-aligned channels across all
    // channels
    for (InputChannelStatus channelStatus : channelStatuses) {
        if (channelStatus.isWatermarkAligned) {
            hasAlignedChannels = true;
            newMinWatermark = Math.min(channelStatus.watermark, newMinWatermark);
        }
    }

    // we acknowledge and output the new overall watermark if it really is aggregated
    // from some remaining aligned channel, and is also larger than the last output watermark
    if (hasAlignedChannels && newMinWatermark > lastOutputWatermark) {
        lastOutputWatermark = newMinWatermark;
        output.emitWatermark(new Watermark(lastOutputWatermark));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

还一个问题也需要探寻一下，那就是在processFunction里，通过ctx.timestamp()获取的时间戳是什么时间？
processionFunction的Context是org.apache.flink.streaming.api.functions.ProcessFunction.Context抽象类，在ProcessOperator中，其默认实现是org.apache.flink.streaming.api.operators.ProcessOperator.ContextImpl，向processFunction内传递的就是ContextImpl对象。
ContextImpl的timestampl()方法实现如下。可以看出，在processFunction内，通过ctx.timestamp()获取到的是StreamRecord的时间戳，而不是系统的watermark。

/**
Timestamp of the element currently being processed or timestamp of a firing timer.
*/
@Override
public Long timestamp() {
    checkState(element != null);

    if (element.hasTimestamp()) {
        return element.getTimestamp();
    } else {
        return null;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

参考材料

1. [白话解析] Flink的Watermark机制
2. 《Flink设计与实现：核心原理与源码解析》
3. 《基于Apache Flink的流处理》
4. Flink WaterMark 详解及实例
5. [源码分析] 从源码入手看 Flink Watermark 之传播过程