Flink 高阶编程：时间语义

本文主要是介绍Flink 高阶编程：时间语义，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

一、时间语义

在 Flink 的流式操作中, 会涉及不同的时间概念

1.1 处理时间

处理时间是指的执行操作的各个设备的时间。

对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟。比如, 一个长度为 1 个小时的窗口将会包含设备时钟表示的 1 个小时内所有的数据. 假设应用程序在 9:15am 分启动, 第 1 个小时窗口将会包含 9:00am 到 10:00am 所有的数据, 然后下个窗口是 10:00am-11:00am, 等等。

处理时间是最简单时间语义, 数据流和设备之间不需要做任何的协调. 他提供了最好的性能和最低的延迟. 但是, 在分布式和异步的环境下, 处理时间没有办法保证确定性, 容易受到数据传递速度的影响: 事件的延迟和乱序。

在使用窗口的时候, 如果使用处理时间, 就指定时间分配器为处理时间分配器。

这篇文章之前所使用的全是处理时间

1.2 事件时间

事件时间是指的这个事件发生的时间.

在 event 进入 Flink 之前, 通常被嵌入到了 event 中, 一般作为这个 event 的时间戳存在.

在事件时间体系中, 时间的进度依赖于数据本身, 和任何设备的时间无关。事件时间程序必须制定如何产生 Event Time Watermarks。在事件时间体系中, Watermarks是表示时间进度的标志(作用就相当于现实时间的时钟)。

在理想情况下，不管事件时间何时到达或者他们的到达的顺序如何, 事件时间处理将产生完全一致且确定的结果。事件时间处理会在等待无序事件(迟到事件)时产生一定的延迟。由于只能等待有限的时间，因此这限制了确定性事件时间应用程序的可使用性。

假设所有数据都已到达，事件时间操作将按预期方式运行，即使在处理无序或迟到的事件或重新处理历史数据时，也会产生正确且一致的结果。例如，每小时事件时间窗口将包含带有事件时间戳的所有记录，该记录落入该小时，无论它们到达的顺序或处理时间。

在使用窗口的时候, 如果使用事件时间, 就指定时间分配器为事件时间分配器。

在 1.12 之前默认的时间语义是处理时间, 从 1.12 开始, Flink 内部已经把默认的语义改成了事件时间

二、WaterMark

考虑这样情况，结合上一篇开窗案例，假设滚动窗口大小 5min，存在一个 9:00-9:05，现在传感器在这个时间段每分钟生成了 5 条数据即 9:00、9:01、9:02、9:03、9:04，但因为网络原因数据在 9:06 才到达 flink 参与计算，但此时 9:00-9:05 窗口已经关闭这批数据就会被丢失。

2.1 原理

事件时间可以不依赖处理时间来表示时间的进度。例如，在程序中即使处理时间和事件时间有相同的速度, 事件时间可能会轻微的落后处理时间。另外一方面，使用事件时间可以在几秒内处理已经缓存在 Kafka 中多周的数据, 这些数据可以照样被正确处理，就像实时发生的一样能够进入正确的窗口。因此在 Flink 中测量事件时间的进度机制就是 WaterMark，同时 WaterMark 作为内嵌在数据流中的特殊数据，并且携带一个时间戳，在开窗中使用作为窗口关闭的衡量标准。具备单调递增的特性。如 9:00-9:05 的窗口，只要 WaterMark 携带的时间没有到 9:05 窗口就不会被关闭。一个 Watermark(t)表示在这个流里面事件时间已经到了时间 t, 意味着此时, 流中不应该存在这样的数据: 他的时间戳 t2<= t (时间比较旧或者等于时间戳)

2.1.1 有序流中的 WaterMark

在下面的这个图中, 事件是有序的(按照他们自己的时间戳来看), WaterMark 是流中一个简单的周期性的标记，通常情况下, WaterMark是一种标记, 是流中的一个点, 所有在这个时间戳(WaterMark中的时间戳)前的数据应该已经全部到达. 一旦 WaterMark 到达了算子, 则这个算子会提高他内部的时钟的值为这个 WaterMark 的值。此时可以解决因延迟迟到的数据。

W(6) 表示事件时间已经到 6s，但实际上内部的最大时间戳是需要减 1ms，类比窗口的结束时间，防止存在两条事件时间相同的数据

2.1.2 无序流中的 WaterMark

若数据不仅有延迟，而且各个数据源延迟不一样，这就导致数据延迟且无序，那么上述的 WaterMark 取法就存在问题，此时需要给予 WaterMark 一个合理的延迟，这个延迟取决去系统最大的乱序程度可以再略高一点

其相当于每个窗口往后延迟一定时间再关闭，数据还是根据自身所携带的时间进入对应的窗口，也就是同一时刻会存在若干个窗口当延迟后的 WaterMark 到达窗口的关闭时间，窗口关闭输出数据。

2.1.3 设定规则

在 Flink 中，WaterMark 由应用程序开发人员生成，这通常需要对相应的领域有一定的了解，如果 WaterMark设置的延迟太久，收到结果的速度可能就会很慢，而如果WaterMark 到达得太早，则可能收到错误结果，不过 Flink 处理迟到数据的机制可以解决这个问题。

2.2 生成器

如何在代码中生成 WaterMark 呢？Flink 内置两个 WaterMark 生成器，forMonotonousTimestamps 和 BoundedOutOfOrdernessWatermarks 分别对应有序和无序流，但从本质上看 forMonotonousTimestamps 是延迟为 0 的 BoundedOutOfOrdernessWatermarks

代码如下：

package day06;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


public class Flink01_WaterMark_EventTime {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度，不设置涉及 watermark 传递
        env.setParallelism(1);

        // 读取数据，转换成 POJO
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("localhost", 1111)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] els = value.split(" ");
                        return new WaterSensor(els[0], Long.parseLong(els[1]), Double.parseDouble(els[2]));
                    }
                }).returns(Types.POJO(WaterSensor.class));

        // 提取事件时间
        WatermarkStrategy<WaterSensor> wm = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        });

        waterSensorDS
                // 注册事件时间
                .assignTimestampsAndWatermarks(wm)
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("vc")
                .print();

        env.execute();
    }
}

结果如下：

只要输入的时间戳小于 1577844005 第一个窗口就不会关闭，此时已经和处理时间没有关系了。

WaterMark 延迟 3s 解决乱序问题，代码如下：

package day06;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;


public class Flink02_WaterMark_OutOfOrder {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度，不设置涉及 watermark 传递
        env.setParallelism(1);

        // 读取数据，转换成 POJO
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("localhost", 1111)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] els = value.split(" ");
                        return new WaterSensor(els[0], Long.parseLong(els[1]), Double.parseDouble(els[2]));
                    }
                }).returns(Types.POJO(WaterSensor.class));

        // 提取事件时间，设置 WaterMark 延迟 3s
        WatermarkStrategy<WaterSensor> wm = WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        return element.getTs() * 1000;
                    }
                });

        waterSensorDS
                // 注册事件时间
                .assignTimestampsAndWatermarks(wm)
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("vc")
                .print();

        env.execute();
    }
}

结果如下：

可以看出即使 1577844001、1577844002 迟到也是可以被统计到的，若是这两条数据在 1577844008 之后到达即超过了延迟时间后才到，那此种方式这两条数据就被丢了。

2.3 多并行度下 WaterMark 传递

注意到上述的代码都是设置并行度为 1，若注释掉并行度或者并行度设置为 2，这时候就会发现即使数据的时间已经到了 1577844008 第一个窗口都不会被关闭，这是因为：多并行度下 WaterMark 向下游传递采用的是轮播，下游取最小的 WaterMark。假设并行度为 2，如图：

此时 WaterMark 到达 8，因为轮播到下游，且下游取最小的导致 keyBy 算子 WaterMark 依然是 1，除非再来一条让 WaterMark 到达 8 的数据，才会让下游的 WaterMark 发生改变。如图：

总结 WaterMark 规律如下：

那么上述问题如何解决呢，设置并行度为 1 肯定不显示，那么可以将生成 WaterMark 的时机提前，在 Source 后直接提取生成即可，代码如下：

package day06;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;


public class Flink03_WaterMark_OutOfOrder2 {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 提取事件时间
        WatermarkStrategy<String> wm = WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        return Long.parseLong(element.split(" ")[1]) * 1000;
                    }
                });

        // 读取数据，转换成 POJO
        env.socketTextStream("localhost", 1111)
                // 注册事件时间
                .assignTimestampsAndWatermarks(wm)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] els = value.split(" ");
                        return new WaterSensor(els[0], Long.parseLong(els[1]), Double.parseDouble(els[2]));
                    }
                }).returns(Types.POJO(WaterSensor.class))
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("vc")
                .print();

        env.execute();
    }
}

再次测试：

2.4 自定义生成器

Flink 提供 2 种风格的 WaterMark 生产方式: periodic(周期性) and punctuated(间歇性)。都需要继承接口: WatermarkGenerator，实际生产中都是 periodic，periodic 默认 200ms 生成一次 WaterMark，punctuated 每条数据生成一次 WaterMark；

优缺点比较：

当数据量大时，punctuated 会产生大量的 WaterMark 占用内存空间
当数据量小时，periodic 会产生大量的 WaterMark，但没有什么影响(Flink 本身就是大数据计算框架，数据量小的时候生成一点 WaterMark 怎么了)

periodic

package day06;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;


public class Flink05_WaterMark_Periodic {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置定时生成时间
        env.getConfig().setAutoWatermarkInterval(200);

        // 设置并行度，不设置涉及 watermark 传递
        env.setParallelism(1);

        // 读取数据，转换成 POJO
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("localhost", 1111)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] els = value.split(" ");
                        return new WaterSensor(els[0], Long.parseLong(els[1]), Double.parseDouble(els[2]));
                    }
                }).returns(Types.POJO(WaterSensor.class));

        // 提取事件时间
        WatermarkStrategy<WaterSensor> wm = new WatermarkStrategy<WaterSensor>() {

            @Override
            public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new MyPeriodic(2000);
            }
        }.withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        });

        SingleOutputStreamOperator<WaterSensor> result = waterSensorDS
                // 注册事件时间
                .assignTimestampsAndWatermarks(wm)
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 允许数据迟到 2s
                .allowedLateness(Time.seconds(2))
                // 3s + 2s 后的数据放入测输出流
                .sideOutputLateData(new OutputTag<WaterSensor>("Side") {
                })
                .sum("vc");

        result.print("master");
        result.getSideOutput(new OutputTag<WaterSensor>("Side") {
        }).print("side");

        env.execute();
    }

    private static class MyPeriodic implements WatermarkGenerator<WaterSensor> {
        private long maxTs;
        // 允许延迟的最大时间
        private final long maxDelay;

        MyPeriodic(long maxDelay) {
            this.maxDelay = maxDelay;
            // 因为周期生成，需要在赋值加 maxDelay，否则调动 onPeriodicEmit 直接变成最大值
            maxTs = Long.MIN_VALUE + maxDelay - 1;
        }

        // 每条数据调用一次
        @Override
        public void onEvent(WaterSensor event, long eventTimestamp, WatermarkOutput output) {
            // 保证 WaterMark 递增
            maxTs = Math.max(maxTs, eventTimestamp);
        }

        // 周期性调用(默认200ms)
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            output.emitWatermark(new Watermark(maxTs - maxDelay - 1));
        }
    }
}

不要问我为什么写的这么漂亮，因为 BoundedOutOfOrdernessWatermarks 就是这么实现的

这篇关于Flink 高阶编程：时间语义的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！

Java教程

Flink 高阶编程：时间语义

一、时间语义

1.1 处理时间

1.2 事件时间

二、WaterMark

2.1 原理

2.1.1 有序流中的 WaterMark

2.1.2 无序流中的 WaterMark

2.1.3 设定规则

2.2 生成器

2.3 多并行度下 WaterMark 传递

2.4 自定义生成器

periodic

前端开发

后端开发

移动端开发

数据库

服务器运维

人工智能

区块链

游戏开发

网站运营

大数据/云计算

软件工程

软件/开发工具使用

资讯