Flink--8、时间语义、水位线（事件和窗口、水位线和窗口的工作原理、生产水位线、水位线的传递、迟到数据的处理）

在这里插入图片描述
星光下的赶路人star的个人主页

将自己生命力展开的人，他的存在，对别人就是愈疗

文章目录

1、时间语义
- 1.1 Flink中的时间语义
- 1.2 哪种时间语义更重要
2、水位线（Watermark）
- 2.1 事件时间和窗口
- 2.2 什么是水位线
- 1.3 水位线和窗口的工作原理
- 1.4 生产水位线
- - 1.4.1 生成水位线的总体原则
  - 1.4.2 水位线生成策略
  - 1.4.3 Flink内置水位线
  - 1.4.4 自定义水位线生成器
- 1.5 水位线的传递
- 1.6 迟到数据的处理
- - 1.6.1 推迟水印推进
  - 1.6.2 设置窗口延迟关闭
  - 1.6.3 使用测流接受迟到的数据

1、时间语义

1.1 Flink中的时间语义

在这里插入图片描述

1.2 哪种时间语义更重要

1、从《星球大战》说起
为了更加清晰地说明两种语义的区别，我们来举一个非常经典的例子：电影《星球大战》。
在这里插入图片描述
如上图所示，我们会发现，看电影其实就是处理影片中数据的过程，所以影片的上映时间就相当于“处理时间”；而影片的数据就是所描述的故事，它所发生的背景时间就相当于“事件时间”。两种时间语义都有各自的用途，适用于不同的场景。
2、数据处理系统中的时间语义
在实际应用中，事件时间语义会更为常见。一般情况下，业务日志数据中都会记录数据生成的时间戳（timestamp），它就可以作为事件时间的判断基础。
在Flink中，由于处理时间比较简单，早期版本默认的时间语义是处理时间；而考虑到事件时间在实际应用中更为广泛，从Flink1.12版本开始，Flink已经将事件时间作为默认的时间语义了。

2、水位线（Watermark）

2.1 事件时间和窗口

在这里插入图片描述

2.2 什么是水位线

在Flink中，用来衡量事件时间进展的标记，就被称作“水位线”（Watermark）。
具体实现上，水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。
在这里插入图片描述
2、乱序流中的水位线

在这里插入图片描述

3、水位线特性

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

水位线是Flink流处理中保证结果正确性的核心机制，它往往会跟窗口一起配合，完成对乱序数据的正确处理。

1.3 水位线和窗口的工作原理

在这里插入图片描述

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

1.4 生产水位线

1.4.1 生成水位线的总体原则

完美的水位线是“绝对正确”的，也就是一个水位线一旦出现就表示这个时间之前的数据已经全部到齐、之后再也不会出现了。不过如果要保证绝对正确，就必须等足够长的时间，这会带来更高的延迟。
如果我们希望处理得更快、实时性更强，那么可以将水位线延迟设得低一些。这种情况下，可能很多迟到数据会在水位线之后才到达，就会导致窗口遗漏数据，计算结果不准确。当然，如果我们对正确性完全不考虑、一味地追求处理速度，可以直接使用处理时间语义，这在理论上可以得到最低的延迟。
所以Flink中的水位线，其实是流处理中对低延迟和结果正确性的一个权衡机制，而且把控制的权利交给了程序员，我们可以在代码中定义水位线的生产策略。

1.4.2 水位线生成策略

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

DataStream<Event> stream = env.addSource(new ClickSource());DataStream<Event> withTimestampsAndWatermarks = 
stream.assignTimestampsAndWatermarks(<watermark strategy>);

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

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

1.4.3 Flink内置水位线

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


/*** 在中间环节产生水印，需要使用：assignTimestampsAndWatermarks(WatermarkStrategy x)** WatermarkStrategy:水印策略。*          包含以下信息：*              （1）水印的特征*                  （a）无水印：watermarkStrategy.noWatermarks()*                  (b)自定义水印特征 WatermarkStrategy.forGenerator()*                  选择系统已经提供：*                      （c）连续水印：数据中提取时间属性-1ms-0*                              WatermarkStrategy.forMonotonousTimeStamps()*                       (d)乱序水印：数据中提前的时间属性-1ms-自定义间隔时间*                               WatermarkStrategy.forBoundedOutOfOrderNess()*              (2)水印的计算方式*                     水印从数据的时间属性中计算得到*                     计算方式的核心功能就是告诉算子，数据中的哪个属性是事件属性*   ----------------------------------------------------------------------------*   一开始玩，一定要把并行度设置为1*/
public class Demo01_ShowWaterMark {public static void main(String[] args) throws Exception {//创建Flink配置类（空参创建的话都是默认值）Configuration configuration = new Configuration();//修改配置类中的WebUI端口号configuration.setInteger("rest.port",3333);//创建Flink环境（并且传入配置对象）StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);//不合并操作算子env.disableOperatorChaining();//设置并行度是1env.setParallelism(1);//自定义水印策略WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps().withTimestampAssigner((e, ts) -> e.getTs());env.socketTextStream("hadoop102",9999).map(new WaterSensorFunction()).assignTimestampsAndWatermarks(watermarkStrategy).keyBy(WaterSensor::getId).process(new KeyedProcessFunction<String, WaterSensor, String>() {@Overridepublic void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {out.collect(value+"="+ctx.timerService().currentWatermark());}}).print();env.execute();}
}

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

public class WatermarkOutOfOrdernessDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 7777).map(new WaterSensorMapFunction());// TODO 1.定义Watermark策略WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy// 1.1 指定watermark生成：乱序的，等待3s.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))// 1.2 指定 时间戳分配器，从数据中提取.withTimestampAssigner((element, recordTimestamp) -> {// 返回的时间戳，要 毫秒System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);return element.getTs() * 1000L;});// TODO 2. 指定 watermark策略SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);sensorDSwithWatermark.keyBy(sensor -> sensor.getId())// TODO 3.使用 事件时间语义 的窗口.window(TumblingEventTimeWindows.of(Time.seconds(10))).process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {long startTs = context.window().getStart();long endTs = context.window().getEnd();String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");long count = elements.spliterator().estimateSize();out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}}).print();env.execute();}
}

测试截图：
在这里插入图片描述

1.4.4 自定义水位线生成器

1、周期性水位生成器（Periodic Generator）
周期性生成器一般是通过onEvent()观察判断输入的事件，而在onPeriodicEmit()里发出水位线。
下面是一段自定义周期性生成水位线的代码：

// 自定义水位线的产生
public class CustomPeriodicWatermarkExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.addSource(new ClickSource()).assignTimestampsAndWatermarks(new CustomWatermarkStrategy()).print();env.execute();}public static class CustomWatermarkStrategy implements WatermarkStrategy<Event> {@Overridepublic TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {return new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event element，long recordTimestamp) {return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段}};}@Overridepublic WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new CustomBoundedOutOfOrdernessGenerator();}}public static class CustomBoundedOutOfOrdernessGenerator implements WatermarkGenerator<Event> {private Long delayTime = 5000L; // 延迟时间private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳@Overridepublic void onEvent(Event event，long eventTimestamp，WatermarkOutput output) {// 每来一条数据就调用一次maxTs = Math.max(event.timestamp，maxTs); // 更新最大时间戳}@Overridepublic void onPeriodicEmit(WatermarkOutput output) {// 发射水位线，默认200ms调用一次output.emitWatermark(new Watermark(maxTs - delayTime - 1L));}}
}

我们在onPeriodicEmit()里调用output.emitWatermark()，就可以发出水位线了；这个方法由系统框架周期性地调用，默认200ms一次。
如果想修改默认周期时间，可以通过下面方法修改。例如：修改为400ms

env.getConfig().setAutoWatermarkInterval(400L);

2、断点式水位生成器（Punctuated Generator）
断点式生成器会不停地检测onEvent()中的事件，当发现带有水位线信息的事件时，就立即发出水位线。我们把发射水位线的逻辑写在onEvent方法当中即可。

3、在数据源中发送水位线
我们也可以在自定义的数据源中抽取事件时间，然后发送水位线。这里要注意的是，在自定义数据源中发送了水位线以后，就不能再在程序中使用assignTimestampsAndWatermarks方法来生成水位线了。在自定义数据源中生成水位线和在程序中使用assignTimestampsAndWatermarks方法生成水位线二者只能取其一。示例程序如下：

env.fromSource(
kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource"
)

1.5 水位线的传递

在这里插入图片描述
在流处理中，上游任务处理完水位线、时钟改变之后，要把当前的水位线再次发出，广播给所以得下游子任务。而当一个任务接受到多个上游并行任务传递来的水位线时，应该以最小的那个作为当前任务的事件时钟。
水位线在上下游任务之间的传递，非常巧妙地避免了分布式系统中没有统一时钟的问题，每个任务都以“处理完之前所有数据”为标准来确定自己的时钟。

1.6 迟到数据的处理

1.6.1 推迟水印推进

在水印产生时，设置一个乱序容忍度，推迟系统时间的推进，保证窗口计算被延迟执行，为乱序的数据争取更多的时间进入窗口。

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10));

1.6.2 设置窗口延迟关闭

Flink的窗口，也允许迟到数据。当触发了窗口计算后，会先计算当前的结果，但是此时并不会关闭窗口。
以后每来一条迟到数据，就触发一次这条数据所在窗口计算(增量计算)。直到wartermark 超过了窗口结束时间+推迟时间，此时窗口会真正关闭。

.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(3))

1.6.3 使用测流接受迟到的数据

.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(3))
.sideOutputLateData(lateWS)

测试代码：

/*** 正常情况下，数据的产生一定是从早到晚。*      早产生的数据，一定是先到达系统的*      很少会有迟到的情况。**      -------------------------------------*      水印的意义：如果当前的算子的水印是x，那么意味着正常情况下，x之前的数据都已经到达了**      迟到：数据的时间属性<水印**      处理迟到的数据:*          迟到的数据，在对应的窗口关闭之前到达，是不受影响的*          迟到的数据，在对应的窗口关闭之后到达，是无法进入窗口的，也无法被计算，此时可以这样处理：*              1、调慢水印时间*                  操作的是水印策略*                  WatermarkStrategy.forBoundedOutOfOrderNess(Duration.ofSeconds(5))**              2、如果1无法解决，可以延迟窗口的关闭时间*                  操作的对象是窗口*                      窗口是到点就计算，然后关闭*                      当我延迟以后，窗口到点就计算，但是不关闭，会延迟一段时间再关闭。在此期间如果有数据，会再次触发窗口计算**              3、如果2无法解决，可以使用测流接受迟到的数据*                  操作的对象是窗口*                  窗口关闭之后的数据，可以导入到一个测流中。后续再处理**                  如果测流中的数据过多，说明当前的系统有问题。排查*                      或数据在产生时，无法保证时序，对于无法保证时序的数据，建议批处理**/
public class Demo03_HandleLate {public static void main(String[] args) throws Exception {//创建Flink配置类（空参创建的话都是默认值）Configuration configuration = new Configuration();//修改配置类中的WebUI端口号configuration.setInteger("rest.port",3333);//创建Flink环境（并且传入配置对象）StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);//设置周期性发送水印(单位是毫秒)env.getConfig().setAutoWatermarkInterval(2000);//关闭算子链env.disableOperatorChaining();//设置并行度env.setParallelism(1);//设置水印特征WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy//设置是乱序水印.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))//从数据中抽取时间戳.withTimestampAssigner((e, ts) -> e.getTs());//设置测流标记OutputTag<WaterSensor> lateTag = new OutputTag<>("late", TypeInformation.of(WaterSensor.class));SingleOutputStreamOperator<String> process = env.socketTextStream("hadoop102", 9999).map(new WaterSensorFunction())//设置水印策略.assignTimestampsAndWatermarks(watermarkStrategy)//全局时间滚动窗口/***  5s的滚动窗口。*                  滚动窗口，从1970-1-1 0：0：0开始当作第一个窗口的起始时间计算。**                  第一个窗口: [0,5000) 等价于 [0,4999]*                              4999就触发第一次运算，但是不关闭。*                              6999关闭*                  第二个窗口: [5000,9999]*                              9999就触发第一次运算，但是不关闭*                              11999关闭*/.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))//运行延迟2s.allowedLateness(Time.seconds(2))//添加测流.sideOutputLateData(lateTag)//运算逻辑.process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {@Overridepublic void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {TimeWindow window = context.window();out.collect(window + ":" + MyUtils.paresToList(elements));}});//主流打印process.print();//测流打印process.getSideOutput(lateTag).printToErr("迟到");//执行env.execute();}
}