一文理解Flink 水位线(Flink Watermark）

r囧r小猫 2024-04-22 06:55 62阅读 0赞

#### 文章目录 ####

*   *  Flink 中的时间语义
     *   *  \`处理时间\`
         *  \`事件时间\`
     *  水位线（Watermark）
     *   *  \`事件时间和窗口\`
         *  \`什么是水位线\`
         *   *  有序流中的水位线
             *  乱序流中的水位线
         *  \`水位线的特性\`
         *  \`如何生成水位线\`
         *   *  水位线生成策略（Watermark Strategies）
             *  有序流、乱序留水位线测试
             *  自定义水位线测试
         *  水位线的传递
         *  水位线的总结

在说水位线之前，我们先了解一下相关的知识，如flink的时间语义中的处理时间、事件时间

### Flink 中的时间语义 ###

对于一台机器而言，“时间”自然就是指系统时间。但我们知道，Flink 是一个分布式处理系统。分布式架构最大的特点，就是节点彼此独立、互不影响，这带来了更高的吞吐量和容错性；但有利必有弊，最大的问题也来源于此。  
在分布式系统中，节点“各自为政”，是没有统一时钟的，数据和控制信息都通过网络进行传输。比如现在有一个任务是窗口聚合，我们希望将每个小时的数据收集起来进行统计处理。而对于并行的窗口子任务，它们所在节点不同，系统时间也会有差异；当我们希望统计 8 点~9 点的数据时，对并行任务来说其实并不是“同时”的，收集到的数据也会有误差。那既然一个集群中有 JobManager 作为管理者，是不是让它统一向所有 TaskManager 发送同步时钟信号就行了呢？这也是不行的。因为网络传输会有延迟，而且这延迟是不确定的，所以 JobManager 发出的同步信号无法同时到达所有节点；想要拥有一个全局统一的时钟，在分布式系统里是做不到的。  
另一个麻烦的问题是，在流式处理的过程中，数据是在不同的节点间不停流动的，这同样  
也会有网络传输的延迟。这样一来，当上下游任务需要跨节点传输数据时，它们对于“时间”的理解也会有所不同。例如，上游任务在 8 点 59 分 59 秒发出一条数据，到下游要做窗口计算时已经是 9 点零 1 秒了，那这条数据到底该不该被收到 8 点~9 点的窗口呢？所以，当我们希望对数据按照时间窗口来进行收集计算时，“时间”到底以谁为标准就非常重要了。  
![在这里插入图片描述][1244bcfbc7c34ed19fc29d3d615f548f.png]  
我们重新梳理一下流式数据处理的过程。如图 6-1 所示，在事件发生之后，生成的数据被收集起来，首先进入分布式消息队列，然后被 Flink 系统中的 Source 算子读取消费，进而向下游的转换算子（窗口算子）传递，最终由窗口算子进行计算处理。  
很明显，这里有两个非常重要的时间点：一个是数据产生的时间，我们把它叫作“事件时间”（Event Time）；另一个是数据真正被处理的时刻，叫作“处理时间”（Processing Time）。我们所定义的窗口操作，到底是以那种时间作为衡量标准，就是所谓的“时间语义”（Notions of Time）。由于分布式系统中网络传输的延迟和时钟漂移，处理时间相对事件发生的时间会有所滞后。

#### `处理时间` ####

处理时间的概念非常简单，就是指执行处理操作的机器的系统时间。  
如果我们以它作为衡量标准，那么数据属于哪个窗口就很明显了：只看窗口任务处理这条数据时，当前的系统时间。比如之前举的例子，数据 8 点 59 分 59 秒产生，而窗口计算时的时间是 9 点零 1 秒，那么这条数据就属于 9 点—10 点的窗口；如果数据传输非常快，9 点之前就到了窗口任务，那么它就属于 8 点—9 点的窗口了。每个并行的窗口子任务，就只按照自己的系统时钟划分窗口。假如我们在早上 8 点 10 分启动运行程序，那么接下来一直到 9 点以前处理的所有数据，都属于第一个窗口；9 点之后、10 点之前的所有数据就将属于第二个窗口。  
这种方法非常简单粗暴，不需要各个节点之间进行协调同步，也不需要考虑数据在流中的位置，简单来说就是“我的地盘听我的”。所以处理时间是最简单的时间语义。

#### `事件时间` ####

事件时间，是指每个事件在对应的设备上发生的时间，也就是数据生成的时间。  
数据一旦产生，这个时间自然就确定了，所以它可以作为一个属性嵌入到数据中。这其实就是这条数据记录的“时间戳”（Timestamp）。  
在事件时间语义下，我们对于时间的衡量，就不看任何机器的系统时间了，而是依赖于数据本身。打个比方，这相当于任务处理的时候自己本身是没有时钟的，所以只好来一个数据就问一下“现在几点了”；而数据本身也没有表，只有一个自带的“出厂时间”，于是任务就基于这个时间来确定自己的时钟。由于流处理中数据是源源不断产生的，一般来说，先产生的数据也会先被处理，所以当任务不停地接到数据时，它们的时间戳也基本上是不断增长的，就可以代表时间的推进。

当然我们会发现，这里有个前提，就是“先产生的数据先被处理”，这要求我们可以保证数据到达的顺序。但是由于分布式系统中网络传输延迟的不确定性，实际应用中我们要面对的数据流往往是乱序的。在这种情况下，就不能简单地把数据自带的时间戳当作时钟了，而需要用另外的标志来表示事件时间进展，在 Flink 中把它叫作事件时间的“水位线”（Watermarks）。

### 水位线（Watermark） ###

#### `事件时间和窗口` ####

事件时间就是事件发生的时间,一个数据产生的时刻，就是流处理中事件触发的时间点，这就是“事件时间”。一个数据产生的时刻，就是流处理中事件触发的时间点，这就是“事件时间”。  
如果我们想要统计一段时间内的数据，需要划分时间窗口，这时只要判断一下时间戳就可以知道数据属于哪个窗口了。  
`举个例子`:一个小时发一趟货车(窗口)，这一个小时任意时间(事件事件)生产的货物都会被装在车上。事件一到就发车(当前窗口关闭),下一趟车继续等待装车发车(下一个窗口运作)。

#### `什么是水位线` ####

在事件时间语义下，我们不依赖系统时间，而是基于数据自带的时间戳去定义了一个时钟，用来表示当前时间的进展。于是每个并行子任务都会有一个自己的逻辑时钟，它的前进是靠数据的时间戳来驱动的。  
但在分布式系统中，这种驱动方式又会有一些问题。因为数据本身在处理转换的过程中会变化，如果遇到窗口聚合这样的操作，其实是要攒一批数据才会输出一个结果，那么下游的数据就会变少，时间进度的控制就不够精细了。另外，数据向下游任务传递时，一般只能传输给一个子任务（除广播外），这样其他的并行子任务的时钟就无法推进了。例如一个时间戳为 9点整的数据到来，当前任务的时钟就已经是 9 点了；处理完当前数据要发送到下游，如果下游任务是一个窗口计算，并行度为 3，那么接收到这个数据的子任务，时钟也会进展到 9 点，9点结束的窗口就可以关闭进行计算了；而另外两个并行子任务则时间没有变化，不能进行窗口计算。  
所以我们应该把时钟也以数据的形式传递出去，告诉下游任务当前时间的进展；而且这个时钟的传递不会因为窗口聚合之类的运算而停滞。一种简单的想法是，在数据流中加入一个时钟标记，记录当前的事件时间；这个标记可以直接广播到下游，当下游任务收到这个标记，就可以更新自己的时钟了。由于类似于水流中用来做标志的记号，在 Flink 中，这种用来衡量事件时间（Event Time）进展的标记，就被称作“水位线”（Watermark）。  
具体实现上，水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。  
![在这里插入图片描述][9b00ff563af042f4bfb27eeab95ad241.png]  
如图 6-5 所示，每个事件产生的数据，都包含了一个时间戳，我们直接用一个整数表示。这里没有指定单位，可以理解为秒或者毫秒（方便起见，下面讲述统一认为是秒）。当产生于2 秒的数据到来之后，当前的事件时间就是 2 秒；在后面插入一个时间戳也为 2 秒的水位线，随着数据一起向下游流动。而当 5 秒产生的数据到来之后，同样在后面插入一个水位线，时间戳也为 5，当前的时钟就推进到了 5 秒。这样，如果出现下游有多个并行子任务的情形，我们只要将水位线广播出去，就可以通知到所有下游任务当前的时间进度了。  
水位线就像它的名字所表达的，是数据流中的一部分，随着数据一起流动，在不同任务之间传输。这看起来非常简单；接下来我们就进一步探讨一些复杂的状况。  
`举个例子`:分布式系统里有很多网络传输或者其他原因导致的传输延迟,如8:59产生的数据,结果9:01到了，如果9点整产生的数据9点整到了，这样当系统在处理的时候，9点产生的数据过来的时候我认为当前事件已经是9点整了(水位线直接变为9点整)，当后面8:59数据过来的时候，认为是迟到的数据，并不是当前时间是8:59。

##### 有序流中的水位线 #####

在理想状态下，数据应该按照它们生成的先后顺序、排好队进入流中；也就是说，它们处理的过程会保持原先的顺序不变，遵守先来后到的原则。这样的话我们从每个数据中提取时间戳，就可以保证总是从小到大增长的，从而插入的水位线也会不断增长、事件时钟不断向前推进。  
实际应用中，如果当前数据量非常大，可能会有很多数据的时间戳是相同的，这时每来一条数据就提取时间戳、插入水位线就做了大量的无用功。而且即使时间戳不同，同时涌来的数据时间差会非常小（比如几毫秒），往往对处理计算也没什么影响。所以为了提高效率，一般会每隔一段时间生成一个水位线，这个水位线的时间戳，就是当前最新数据的时间戳，如图6-6 所示。所以这时的水位线，其实就是有序流中的一个周期性出现的时间标记。  
![在这里插入图片描述][cd6d256bc8d14476bc8fd2b475b5773c.png]

##### 乱序流中的水位线 #####

有序流的处理非常简单，看起来水位线也并没有起到太大的作用。但这种情况只存在于理想状态下。我们知道在分布式系统中，数据在节点间传输，会因为网络传输延迟的不确定性，导致顺序发生改变，这就是所谓的“乱序数据”。  
这里所说的“乱序”（out-of-order），是指数据的先后顺序不一致，主要就是基于数据的产生时间而言的。如图 6-7 所示，一个 7 秒时产生的数据，生成时间自然要比 9 秒的数据早；但是经过数据缓存和传输之后，处理任务可能先收到了 9 秒的数据，之后 7 秒的数据才姗姗来迟(迟到数据)。  
![在这里插入图片描述][35eb42eb6bf04be39dae320df61bf038.png]  
最直观的想法自然是跟之前一样，我们还是靠数据来驱动，每来一个数据就提取它的时间戳、插入一个水位线。不过现在的情况是数据乱序，所以有可能新的时间戳比之前的还小，如果直接将这个时间的水位线再插入，我们的“时钟”就回退了——水位线就代表了时钟，时光不能倒流，所以水位线的时间戳也不能减小。  
解决思路也很简单：我们插入新的水位线时，要先判断一下时间戳是否比之前的大，否则就不再生成新的水位线，如图 6-8 所示。也就是说，只有数据的时间戳比当前时钟大，才能推动时钟前进，这时才插入水位线。  
![在这里插入图片描述][7fb61df17d104905ae60f5a7bd38e5bb.png]  
这样做尽管可以定义出一个事件时钟，却也会带来一个非常大的问题：我们无法正确处理“迟到”的数据。在上面的例子中，当 9 秒产生的数据到来之后，我们就直接将时钟推进到了9 秒；如果有一个窗口结束时间就是 9 秒（比如，要统计 0~9 秒的所有数据），那么这时窗口就应该关闭、将收集到的所有数据计算输出结果了。但事实上，由于数据是乱序的，还可能有时间戳为 7 秒、8 秒的数据在 9 秒的数据之后才到来，这就是“迟到数据”（late data）。它们本来也应该属于 0~9 秒这个窗口，但此时窗口已经关闭，于是这些数据就被遗漏了，这会导致统计结果不正确。  
为了让窗口能够正确收集到迟到的数据，我们也可以等上 2 秒；也就是用当前已有数据的最大时间戳减去 2 秒，就是要插入的水位线的时间戳，如图 6-10 所示。这样的话，9 秒的数据到来之后，事件时钟不会直接推进到 9 秒，而是进展到了 7 秒；必须等到11 秒的数据到来之后，事件时钟才会进展到 9 秒，这时迟到数据也都已收集齐，0~9 秒的窗口就可以正确计算结果了。  
![在这里插入图片描述][60578234e0874ee1971dee02af34118e.png]  
如果仔细观察就会看到，这种“等 2 秒”的策略其实并不能处理所有的乱序数据。比如22 秒的数据到来之后，插入的水位线时间戳为 20，也就是当前时钟已经推进到了 20 秒；对于10~20 秒的窗口，这时就该关闭了。但是之后又会有 17 秒的迟到数据到来，它本来应该属于10~20 秒窗口，现在却被遗漏丢弃了。那又该怎么办呢？  
既然现在等 2 秒还是等不到 17 秒产生的迟到数据，那自然我们可以试着多等几秒，也就  
是把时钟调得更慢一些。最终的目的，就是要让窗口能够把所有迟到数据都收进来，得到正确的计算结果。对应到水位线上，其实就是要保证，当前时间已经进展到了这个时间戳，在这之后不可能再有迟到数据来了。  
下面是一个示例，我们可以使用周期性的方式生成正确的水位线。  
![在这里插入图片描述][73d8c9be8b2a4960a39693770de0ac9a.png]  
如图 6-11 所示，第一个水位线时间戳为 7，它表示当前事件时间是 7 秒，7 秒之前的数据都已经到齐，之后再也不会有了；同样，第二个、第三个水位线时间戳分别为 12 和 20，表示11 秒、20 秒之前的数据都已经到齐，如果有对应的窗口就可以直接关闭了，统计的结果一定是正确的。这里由于水位线是周期性生成的，所以插入的位置不一定是在时间戳最大的数据后面。

#### `水位线的特性` ####

⚫ 水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据  
⚫ 水位线主要的内容是一个时间戳，用来表示当前事件时间的进展  
⚫ 水位线是基于数据的时间戳生成的  
⚫ 水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进  
⚫ 水位线可以通过设置延迟，来保证正确处理乱序数据  
⚫ 一个水位线 Watermark(t)，表示在当前流中事件时间已经达到了时间戳 t, 这代表 t 之前的所有数据都到齐了，之后流中不会出现时间戳 t’ ≤ t 的数据

#### `如何生成水位线` ####

一个字，等。由于网络传输的延迟不确定，为了获取所有迟到数据，我们只能等待更长的时间。作为筹划全局的程序员，我们当然不会傻傻地一直等下去。那到底等多久呢？这就需要对相关领域有一定的了解了。比如，如果我们知道当前业务中事件的迟到时间不会超过 5 秒，那就可以将水位线的时间戳设为当前已有数据的最大时间戳减去 5 秒，相当于设置了 5 秒的延迟等待。  
更多的情况下，我们或许没那么大把握。毕竟未来是没有人能说得准的，我们怎么能确信未来不会出现一个超级迟到数据呢？所以另一种做法是，可以单独创建一个 Flink 作业来监控事件流，建立概率分布或者机器学习模型，学习事件的迟到规律。  
如果我们希望计算结果能更加准确，那可以将水位线的延迟设置得更高一些，等待的时间越长，自然也就越不容易漏掉数据。不过这样做的代价是处理的实时性降低了，我们可能为极少数的迟到数据增加了很多不必要的延迟。

##### 水位线生成策略（Watermark Strategies） #####

在 Flink 的 DataStream API 中 ， 有 一 个 单 独 用 于 生 成 水 位 线 的 方法：.assignTimestampsAndWatermarks()，它主要用来为流中的数据分配时间戳，并生成水位线来指示事件时间

**代码示例**

<dependencies>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>${slf4j.version}</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>${slf4j.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-to-slf4j</artifactId>
                <version>2.14.0</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.22</version>
                <scope>compile</scope>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.58</version>
            </dependency>
    
        
        <dependency>
            <groupId>org.apache.bahir</groupId>
            <artifactId>flink-connector-redis_2.11</artifactId>
            <version>1.0</version>
        </dependency>
    
        
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
    
         
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.47</version>
        </dependency>
    
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>2.2.4</version>
        </dependency>
        </dependencies>

import com.event.UserEvent;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    import java.util.Calendar;
    import java.util.Random;
    
    public class CustomSouce implements SourceFunction<UserEvent> {
        // 声明一个布尔变量，作为控制数据生成的标识位
        private Boolean running = true;
    
        @Override
        public void run(SourceContext<UserEvent> ctx) throws Exception {
            Random random = new Random(); // 在指定的数据集中随机选取数据
            String[] users = {"Mary", "Alice", "Bob", "Cary"};
            String[] urls = {"./home", "./cart", "./fav", "./prod?id=1",
                    "./prod?id=2"};
            while (running) {
                ctx.collect(new UserEvent(
                        users[random.nextInt(users.length)],
                        urls[random.nextInt(urls.length)],
                        Calendar .getInstance().getTimeInMillis()
                ));
                // 隔 1 秒生成一个点击事件，方便观测
                Thread.sleep(1000);
            }
        }
        @Override
        public void cancel() {
            running = false;
        }
    }

import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import lombok.ToString;
    
    @Data
    @NoArgsConstructor
    @ToString
    @AllArgsConstructor
    public class UserEvent {
        
        private String userName;
        private String url;
        private Long timestemp;
    }

##### 有序流、乱序留水位线测试 #####

import com.event.UserEvent;
    import my.test.source.CustomSouce;
    import org.apache.flink.api.common.eventtime.*;
    import org.apache.flink.shaded.jackson2.org.yaml.snakeyaml.events.Event;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.time.Duration;
    import java.util.Arrays;
    
    public class WaterMarkTest {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            env.getConfig().setAutoWatermarkInterval(100); //水位线之间的间隔（以毫秒为单位）。
    
            DataStream<UserEvent> userEventDataStreamSource =
                    env.addSource(new CustomSouce())
                            //有序流测试
    //                        .assignTimestampsAndWatermarks(
    //                                WatermarkStrategy.<UserEvent>forMonotonousTimestamps()
    //                                .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
    //                                    @Override
    //                                    public long extractTimestamp(UserEvent element, long recordTimestamp) {
    //
    //                                        return element.getTimestemp();
    //                                    }
    //                                }))
                        //乱序流中水位线测试
                        .assignTimestampsAndWatermarks(
                        // 针对乱序流插入水位线，延迟时间设置为 100ms
                        WatermarkStrategy.
     <UserEvent>forBoundedOutOfOrderness(Duration.ofMillis(100)).withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
                            @Override
                            public long extractTimestamp(UserEvent element, long recordTimestamp) {
                               // 抽取时间戳的逻辑
                                return element.getTimestemp();
                            }
                        }));
    
                userEventDataStreamSource.print();
                env.execute();
        }
    
    
    }

##### 自定义水位线测试 #####

import com.event.UserEvent;
    import my.test.source.CustomSouce;
    import org.apache.flink.api.common.eventtime.*;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.time.Duration;
    import java.util.Arrays;
    
    public class CustomWatermarkTest {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env =
                    StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.addSource(new CustomSouce())
                .assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
                .print();
            env.execute();
        }
        public static class CustomWatermarkStrategy implements
                WatermarkStrategy<UserEvent> {
            @Override
            public TimestampAssigner<UserEvent>
            createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                return new SerializableTimestampAssigner<UserEvent>() {
                    @Override
                    public long extractTimestamp(UserEvent element, long recordTimestamp)
                    {
                        return element.getTimestemp(); // 告诉程序数据源里的时间戳是哪一个字段
                    }
                };
            }
            @Override
            public WatermarkGenerator<UserEvent>
            createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new CustomPeriodicGenerator();
            }
        }
        public static class CustomPeriodicGenerator implements
                WatermarkGenerator<UserEvent> {
            private Long delayTime = 5000L; // 延迟时间
            private Long maxTs = Long.MIN_VALUE + delayTime + 1L; // 观察到的最大时间戳
            @Override
            public void onEvent(UserEvent event, long eventTimestamp, WatermarkOutput
                    output) {
                // 每来一条数据就调用一次
                maxTs = Math.max(event.getTimestemp(), maxTs); // 更新最大时间戳
            }
            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                // 发射水位线，默认 200ms 调用一次
                output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
            }
        }
    }

#### 水位线的传递 ####

我们知道水位线是数据流中插入的一个标记，用来表示事件时间的进展，它会随着数据一起在任务间传递。如果只是直通（forward）的传输，那很简单，数据和水位线都是按照本身的顺序依次传递、依次处理的；一旦水位线到达了算子任务, 那么这个任务就会将它内部的时钟设为这个水位线的时间戳。  
在这里，“任务的时钟”其实仍然是各自为政的，并没有统一的时钟。实际应用中往往上下游都有多个并行子任务，为了统一推进事件时间的进展，我们要求上游任务处理完水位线、时钟改变之后，要把当前的水位线再次发出，广播给所有的下游子任务。这样，后续任务就不需要依赖原始数据中的时间戳（经过转化处理后，数据可能已经改变了），也可以知道当前事件时间了。  
![在这里插入图片描述][8102c0b7472c489692e6513e110a1aae.png]  
当有一个新的水位线（第一分区的 4）从上游传来时，当前任务会首先更新对应的分区时钟；然后再次判断所有分区时钟中的最小值，如果比之前大，说明事件时间有了进展，当前任务的时钟也就可以更新了。这里要注意，更新后的任务时钟，并不一定是新来的那个分区水位线，比如这里改变的是第一分区的时钟，但最小的分区时钟是第三分区的 3，于是当前任务时钟就推进到了 3。当时钟有进展时，当前任务就会将自己的时钟以水位线的形式，广播给  
下游所有子任务。  
再次收到新的水位线（第二分区的 7）后，执行同样的处理流程。首先将第二个分区时钟更新为 7，然后比较所有分区时钟；发现最小值没有变化，那么当前任务的时钟也不变，也不会向下游任务发出水位线。  
同样道理，当又一次收到新的水位线（第三分区的 6）之后，第三个分区时钟更新为6，同时所有分区时钟最小值变成了第一分区的 4，所以当前任务的时钟推进到 4，并发出时间戳为 4 的水位线，广播到下游各个分区任务。  
水位线在上下游任务之间的传递，非常巧妙地避免了分布式系统中没有统一时钟的问题，每个任务都以“处理完之前所有数据”为标准来确定自己的时钟，就可以保证窗口处理的结果总是正确的。对于有多条流合并之后进行处理的场景，水位线传递的规则是类似的。

#### 水位线的总结 ####

水位线在事件时间的世界里面，承担了时钟的角色。也就是说在事件时间的流中，水位线是唯一的时间尺度。如果想要知道现在几点，就要看水位线的大小。后面讲到的窗口的闭合，以及定时器的触发都要通过判断水位线的大小来决定是否触发。水位线是一种特殊的事件，由程序员通过编程插入的数据流里面，然后跟随数据流向下游流动。  
`水位线的默认计算公式：水位线 = 观察到的最大事件时间 – 最大延迟时间 – 1 毫秒。`  
所以这里涉及到一个问题，就是不同的算子看到的水位线的大小可能是不一样的。因为下游的算子可能并未接收到来自上游算子的水位线，导致下游算子的时钟要落后于上游算子的时钟。比如 map->reduce 这样的操作，如果在 map 中编写了非常耗时间的代码，将会阻塞水位线的向下传播，因为水位线也是数据流中的一个事件，位于水位线前面的数据如果没有处理完毕，那么水位线不可能弯道超车绕过前面的数据向下游传播，也就是说会被前面的数据阻塞。这样就会影响到下游算子的聚合计算，因为下游算子中无论由窗口聚合还是定时器的操作，都需要水位线才能触发执行。这也就告诉了我们，在编写 Flink 程序时，一定要谨慎的编写每一个算子的计算逻辑，尽量避免大量计算或者是大量的 IO 操作，这样才不会阻塞水位线的向下传递。  
在数据流开始之前，Flink 会插入一个大小是负无穷大（在 Java 中是-Long.MAX\_VALUE）的水位线，而在数据流结束时，Flink 会插入一个正无穷大(Long.MAX\_VALUE)的水位线，保证所有的窗口闭合以及所有的定时器都被触发。对于离线数据集，Flink 也会将其作为流读入，也就是一条数据一条数据的读取。在这种情况下，Flink 对于离线数据集，只会插入两次水位线，也就是在最开始处插入负无穷大的水位线，在结束位置插入一个正无穷大的水位线。因为只需要插入两次水位线，就可以保证计算的正确，无需在数据流的中间插入水位线了。水位线的重要性在于它的逻辑时钟特性，而逻辑时钟这个概念可以说是分布式系统里面最为重要的概念之一了，理解透彻了对理解各种分布式系统非常有帮助。具体可以参考 LeslieLamport 的论文。

参考视频地址:https://www.bilibili.com/video/BV133411s7Sa/?vd\_source=71b8b6ac6bbf46ba0a21b458e5e32aa0

[1244bcfbc7c34ed19fc29d3d615f548f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/c541bf9cd6e448efaa6b6a73c47ceb7b.png
[9b00ff563af042f4bfb27eeab95ad241.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/7bdc527fef8a433594fb9965224fce23.png
[cd6d256bc8d14476bc8fd2b475b5773c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/845e17d0d8074aa9b5aba89e355c2b57.png
[35eb42eb6bf04be39dae320df61bf038.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/38b7ab8d209d4959bf10823be4f4c52c.png
[7fb61df17d104905ae60f5a7bd38e5bb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/5430a316a8f342a3a3ada41a618ea202.png
[60578234e0874ee1971dee02af34118e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/3dabea7c31244554915edcb690025ab3.png
[73d8c9be8b2a4960a39693770de0ac9a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/3278611579f64927bad8c2c29aa32113.png
[8102c0b7472c489692e6513e110a1aae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/f1e2f8ee8eb14393819ff07a34beb6e5.png