案例用到的测试数据请参考文章：

Flink自定义Source模拟数据流

原文链接：https://blog.csdn.net/m0_52606060/article/details/135436048

Flink中的时间语义

哪种时间语义更重要

从《星球大战》说起

数据处理系统中的时间语义

在实际应用中，事件时间语义会更为常见。一般情况下，业务日志数据中都会记录数据生成的时间戳（timestamp），它就可以作为事件时间的判断基础。

在Flink中，由于处理时间比较简单，早期版本默认的时间语义是处理时间；而考虑到事件时间在实际应用中更为广泛，从Flink1.12版本开始，Flink已经将事件时间作为默认的时间语义了。

水位线（Watermark）

事件时间和窗口

什么是水位线

在Flink中，用来衡量事件时间进展的标记，就被称作“水位线”（Watermark）。

具体实现上，水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。

水位线和窗口的工作原理

注意：Flink中窗口并不是静态准备好的，而是动态创建——当有落在这个窗口区间范围的数据达到时，才创建对应的窗口。另外，这里我们认为到达窗口结束时间时，窗口就触发计算并关闭，事实上“触发计算”和“窗口关闭”两个行为也可以分开，这部分内容我们会在后面详述。

生成水位线

生成水位线的总体原则

完美的水位线是“绝对正确”的，也就是一个水位线一旦出现，就表示这个时间之前的数据已经全部到齐、之后再也不会出现了。不过如果要保证绝对正确，就必须等足够长的时间，这会带来更高的延迟。

如果我们希望处理得更快、实时性更强，那么可以将水位线延迟设得低一些。这种情况下，可能很多迟到数据会在水位线之后才到达，就会导致窗口遗漏数据，计算结果不准确。当然，如果我们对准确性完全不考虑、一味地追求处理速度，可以直接使用处理时间语义，这在理论上可以得到最低的延迟。

所以Flink中的水位线，其实是流处理中对低延迟和结果正确性的一个权衡机制，而且把控制的权力交给了程序员，我们可以在代码中定义水位线的生成策略。

水位线生成策略

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

DataStream stream = env.addSource(new ClickSource());
DataStream withTimestampsAndWatermarks = 
stream.assignTimestampsAndWatermarks();

说明：WatermarkStrategy作为参数，这就是所谓的“水位线生成策略”。WatermarkStrategy是一个接口，该接口中包含了一个“时间戳分配器”TimestampAssigner和一个“水位线生成器”

WatermarkGenerator。
public interface WatermarkStrategy 
    extends TimestampAssignerSupplier,
            WatermarkGeneratorSupplier{
    // 负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。
    @Override
    TimestampAssigner createTimestampAssigner(TimestampAssignerSupplier.Context context);
    // 主要负责按照既定的方式，基于时间戳生成水位线
    @Override
    WatermarkGenerator createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

Flink内置水位线

有序流中内置水位线设置

对于有序流，主要特点就是时间戳单调增长，所以永远不会出现迟到数据的问题。这是周期性生成水位线的最简单的场景，直接调用WatermarkStrategy.forMonotonousTimestamps()方法就可以实现。

import com.zxl.bean.Orders;
import com.zxl.datas.OrdersData;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
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 org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class WatermarkStrategyDemo {
    public static void main(String[] args) throws Exception {
        //创建Flink流处理执行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        environment.setParallelism(1);
        // TODO: 2024/1/7 定义时间语义
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //调用Flink自定义Source
        // TODO: 2024/1/6 订单数据
        DataStreamSource ordersDataStreamSource = environment.addSource(new OrdersData());
        // TODO: 2024/1/11 定义Watermark策略
        WatermarkStrategy ordersWatermarkStrategy = WatermarkStrategy.forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner() {
                    @Override
                    public long extractTimestamp(Orders orders, long l) {
                        System.out.println("返回时间戳"+orders.getOrder_date()+"毫秒13位的");
                        return orders.getOrder_date();
                    }
                });
        // TODO: 2024/1/11  指定 watermark策略
        SingleOutputStreamOperator watermarks = ordersDataStreamSource.assignTimestampsAndWatermarks(ordersWatermarkStrategy);
        // TODO: 2024/1/11 进行聚合运算
        SingleOutputStreamOperator reduce = watermarks.keyBy(orders -> orders.getProduct_id())
                // TODO: 2024/1/11 定义时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction() {
                    @Override
                    public Orders reduce(Orders orders, Orders t1) throws Exception {
                        Orders orders1 = new Orders(t1.getOrder_id(), t1.getUser_id(), t1.getOrder_date(), t1.getOrder_amount() + orders.getOrder_amount(), t1.getProduct_id(), t1.getOrder_num());
                        return orders1;
                    }
                });
        ordersDataStreamSource.print("聚合前数据");
        reduce.print("聚合数据");
        environment.execute();
    }
}

乱序流中内置水位线设置

由于乱序流中需要等待迟到数据到齐，所以必须设置一个固定量的延迟时间。这时生成水位线的时间戳，就是当前数据流中最大的时间戳减去延迟的结果，相当于把表调慢，当前时钟会滞后于数据的最大时间戳。调用WatermarkStrategy. forBoundedOutOfOrderness()方法就可以实现。这个方法需要传入一个maxOutOfOrderness参数，表示“最大乱序程度”，它表示数据流中乱序数据时间戳的最大差值；如果我们能确定乱序程度，那么设置对应时间长度的延迟，就可以等到所有的乱序数据了。

import com.zxl.bean.Orders;
import com.zxl.datas.OrdersData;
import javafx.scene.input.DataFormat;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
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 org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple2;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
public class WatermarkOutOfOrdernessDemo {
    public static void main(String[] args) throws Exception {
        //创建Flink流处理执行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        environment.setParallelism(1);
        // TODO: 2024/1/7 定义时间语义
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //调用Flink自定义Source
        // TODO: 2024/1/6 订单数据
        DataStreamSource ordersDataStreamSource = environment.addSource(new OrdersData());
        // TODO: 2024/1/11 定义Watermark策略
        WatermarkStrategy ordersWatermarkStrategy = WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3))
                // TODO: 2024/1/11 指定水位线时间戳也可以用 Lambda 表达式
                .withTimestampAssigner((orders, l) -> orders.getOrder_date());
        // TODO: 2024/1/11 指定水位线
        SingleOutputStreamOperator operator = ordersDataStreamSource.assignTimestampsAndWatermarks(ordersWatermarkStrategy);
        // TODO: 2024/1/11 分组聚合
        SingleOutputStreamOperator