FlinkAPI开发之水位线（Watermark）

2024-01-16 16:12
文章标签 开发 水位 watermark flinkapi

本文主要是介绍FlinkAPI开发之水位线（Watermark），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

案例用到的测试数据请参考文章：
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<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);
}

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();//设置并行度为1environment.setParallelism(1);// TODO: 2024/1/7 定义时间语义environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//调用Flink自定义Source// TODO: 2024/1/6 订单数据DataStreamSource<Orders> ordersDataStreamSource = environment.addSource(new OrdersData());// TODO: 2024/1/11 定义Watermark策略WatermarkStrategy<Orders> ordersWatermarkStrategy = WatermarkStrategy.<Orders>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<Orders>() {@Overridepublic long extractTimestamp(Orders orders, long l) {System.out.println("返回时间戳"+orders.getOrder_date()+"毫秒13位的");return orders.getOrder_date();}});// TODO: 2024/1/11  指定 watermark策略SingleOutputStreamOperator<Orders> watermarks = ordersDataStreamSource.assignTimestampsAndWatermarks(ordersWatermarkStrategy);// TODO: 2024/1/11 进行聚合运算SingleOutputStreamOperator<Orders> reduce = watermarks.keyBy(orders -> orders.getProduct_id())// TODO: 2024/1/11 定义时间窗口.window(TumblingEventTimeWindows.of(Time.seconds(5))).reduce(new ReduceFunction<Orders>() {@Overridepublic 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();//设置并行度为1environment.setParallelism(1);// TODO: 2024/1/7 定义时间语义environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//调用Flink自定义Source// TODO: 2024/1/6 订单数据DataStreamSource<Orders> ordersDataStreamSource = environment.addSource(new OrdersData());// TODO: 2024/1/11 定义Watermark策略WatermarkStrategy<Orders> ordersWatermarkStrategy = WatermarkStrategy.<Orders>forBoundedOutOfOrderness(Duration.ofSeconds(3))// TODO: 2024/1/11 指定水位线时间戳也可以用 Lambda 表达式.withTimestampAssigner((orders, l) -> orders.getOrder_date());// TODO: 2024/1/11 指定水位线SingleOutputStreamOperator<Orders> operator = ordersDataStreamSource.assignTimestampsAndWatermarks(ordersWatermarkStrategy);// TODO: 2024/1/11 分组聚合SingleOutputStreamOperator<Object> reduce = operator.keyBy(orders -> orders.getProduct_id()).window(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessWindowFunction<Orders, Object, Integer, TimeWindow>() {// TODO: 2024/1/9 参数说明：分组key值，窗口计时器,按照key分组后的数据，收集器@Overridepublic void process(Integer integer, ProcessWindowFunction<Orders, Object, Integer, TimeWindow>.Context context, Iterable<Orders> elements, Collector<Object> collector) throws Exception {// TODO: 2024/1/9 窗口内同一个key包含的数据条数long count = elements.spliterator().estimateSize();// TODO: 2024/1/9 窗口的开始时间long windowStartTs = context.window().getStart();// TODO: 2024/1/9 窗口的结束时间long windowEndTs = context.window().getEnd();String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");// TODO: 2024/1/9 输出收集器collector.collect("key=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}});ordersDataStreamSource.map(new MapFunction<Orders, Tuple2<String,Integer>>() {@Overridepublic Tuple2<String, Integer> map(Orders orders) throws Exception {//时间格式,HH是24小时制，hh是AM PM12小时制SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//比如timestamp=1449210225945；String date_string = sdf.format(new Date(orders.getOrder_date()));return new Tuple2<>(date_string,orders.getOrder_amount());}}).print();reduce.print("聚合数据");environment.execute();}
}

在这里插入图片描述

自定义水位线生成器

周期性水位线生成器（Periodic Generator）

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

import com.zxl.bean.Orders;
import com.zxl.datas.OrdersData;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
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.util.Date;public class CustomPeriodicWatermarkExample {public static void main(String[] args) throws Exception {//创建Flink流处理执行环境StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();//设置并行度为1environment.setParallelism(1);// TODO: 2024/1/7 定义时间语义environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//调用Flink自定义Source// TODO: 2024/1/6 订单数据DataStreamSource<Orders> ordersDataStreamSource = environment.addSource(new OrdersData());// TODO: 2024/1/12 设置水位线 SingleOutputStreamOperator<Orders> operator = ordersDataStreamSource.assignTimestampsAndWatermarks(new CustomWatermarkStrategy());// TODO: 2024/1/12 打印数据ordersDataStreamSource.map(new MapFunction<Orders, Tuple2<String,Integer>>() {@Overridepublic Tuple2<String, Integer> map(Orders orders) throws Exception {//时间格式,HH是24小时制，hh是AM PM12小时制SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//比如timestamp=1449210225945；String date_string = sdf.format(new Date(orders.getOrder_date()));return new Tuple2<>(date_string,orders.getOrder_amount());}}).print();// TODO: 2024/1/11 分组聚合SingleOutputStreamOperator<Object> reduce = operator.keyBy(orders -> orders.getProduct_id()).window(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessWindowFunction<Orders, Object, Integer, TimeWindow>() {// TODO: 2024/1/9 参数说明：分组key值，窗口计时器,按照key分组后的数据，收集器@Overridepublic void process(Integer integer, ProcessWindowFunction<Orders, Object, Integer, TimeWindow>.Context context, Iterable<Orders> elements, Collector<Object> collector) throws Exception {// TODO: 2024/1/9 窗口内同一个key包含的数据条数long count = elements.spliterator().estimateSize();// TODO: 2024/1/9 窗口的开始时间long windowStartTs = context.window().getStart();// TODO: 2024/1/9 窗口的结束时间long windowEndTs = context.window().getEnd();String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");// TODO: 2024/1/9 输出收集器collector.collect("key=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}});reduce.print("聚合数据");environment.execute();}// TODO: 2024/1/12 自定义水位线生成器public static class CustomWatermarkStrategy implements WatermarkStrategy<Orders>{@Overridepublic WatermarkGenerator<Orders> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new WatermarkGenerator<Orders>() {// TODO: 2024/1/12  延迟时间private Long delayTime = 2000L;// TODO: 2024/1/12  观察到的最大时间戳private Long maxTs = -Long.MAX_VALUE + delayTime + 1L;@Overridepublic void onEvent(Orders orders, long l, WatermarkOutput watermarkOutput) {// TODO: 2024/1/12   每来一条数据就调用一次 , 更新最大时间戳maxTs = Math.max(orders.getOrder_date(),maxTs);}@Overridepublic void onPeriodicEmit(WatermarkOutput output) {// TODO: 2024/1/12   发射水位线，默认200ms调用一次output.emitWatermark(new Watermark(maxTs - delayTime - 1L));}};}@Overridepublic TimestampAssigner<Orders> createTimestampAssigner(TimestampAssignerSupplier.Context context) {// TODO: 2024/1/12  告诉程序数据源里的时间戳是哪一个字段 return new SerializableTimestampAssigner<Orders>(){@Overridepublic long extractTimestamp(Orders orders, long l) {return orders.getOrder_date();}};}}
}

在这里插入图片描述

断点式水位线生成器（Punctuated Generator）

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

在数据源中发送水位线

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

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

水位线的传递

在这里插入图片描述

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

案例：乱序流的watermark，将并行度设为2，观察现象。
在多个上游并行任务中，如果有其中一个没有数据，由于当前Task是以最小的那个作为当前任务的事件时钟，就会导致当前Task的水位线无法推进，就可能导致窗口无法触发。这时候可以设置空闲等待。

import com.zxl.bean.Orders;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
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 java.time.Duration;public class WatermarkOutOfOrdernessDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();environment.setParallelism(2);DataStream<String> socketTextStream = environment.socketTextStream("175.24.186.230", 9999);DataStream<Integer> streamOperator = socketTextStream.map(Integer::parseInt);//自定义分区DataStream<Integer> dataStream = streamOperator.partitionCustom(new Partitioner<Integer>() {@Overridepublic int partition(Integer integer, int i) {if (integer % 2 == 0) {return 0;} else {return 1;}}}, new KeySelector<Integer, Integer>() {@Overridepublic Integer getKey(Integer integer) throws Exception {return integer;}});WatermarkStrategy<Orders> ordersWatermarkStrategy = WatermarkStrategy.<Orders>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((orders, l) -> orders.getOrder_date());SingleOutputStreamOperator<Integer> watermarks = dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.<Integer>forMonotonousTimestamps().withTimestampAssigner((r, ts) -> r * 1000L)// TODO: 2024/1/12 空闲等待5s.withIdleness(Duration.ofSeconds(5)));// 分成两组： 奇数一组，偶数一组 ， 开10s的事件时间滚动窗口watermarks.keyBy(r -> r % 2).window(TumblingEventTimeWindows.of(Time.seconds(10))).process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {@Overridepublic void process(Integer integer, Context context, Iterable<Integer> 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=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}}).print();environment.execute();}
}

在这里插入图片描述

迟到数据的处理

推迟水印推进

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

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

设置窗口延迟关闭

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

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

注意:允许迟到只能运用在event time上

import com.zxl.bean.Orders;
import com.zxl.datas.OrdersData;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
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 org.apache.flink.util.OutputTag;import java.time.Duration;public class WatermarkLateDemo {public static void main(String[] args) throws Exception {// TODO: 2024/1/15  创建Flink流处理执行环境StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();// TODO: 2024/1/15  设置并行度为1environment.setParallelism(1);// TODO: 2024/1/7 定义时间语义environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);// TODO: 2024/1/6 订单数据DataStreamSource<Orders> ordersDataStreamSource = environment.addSource(new OrdersData());// TODO: 2024/1/15 设置水位线WatermarkStrategy<Orders> watermarkStrategy = WatermarkStrategy// TODO: 2024/1/15 设置最大乱序程度，数据流中乱序数据时间戳的最大差值.<Orders>forBoundedOutOfOrderness(Duration.ofSeconds(3))// TODO: 2024/1/15 指定水位线中对应的时间字段 .withTimestampAssigner((orders, l) -> orders.getOrder_date())// TODO: 2024/1/15 设置空闲等待时间，如果某个分区中有长时间无数据产生将放弃此分区的水位线选举权力 .withIdleness(Duration.ofSeconds(3));// TODO: 2024/1/15 添加水位线SingleOutputStreamOperator<Orders> operator = ordersDataStreamSource.assignTimestampsAndWatermarks(watermarkStrategy);// TODO: 2024/1/15  定义侧输出流的标签OutputTag<Orders> lateData = new OutputTag<>("late_data", Types.POJO(Orders.class));// TODO: 2024/1/15 设置分区keySingleOutputStreamOperator<Object> process = operator.keyBy(orders -> orders.getProduct_id())// TODO: 2024/1/15 定义滚动窗口时间大小.window(TumblingEventTimeWindows.of(Time.seconds(5)))// TODO: 2024/1/15 窗口延迟关闭时间.allowedLateness(Time.seconds(3))// TODO: 2024/1/15 迟到数据的测输出流.sideOutputLateData(lateData)// TODO: 2024/1/15 进行数据聚合.process(new ProcessWindowFunction<Orders, Object, Integer, TimeWindow>() {@Overridepublic void process(Integer integer, ProcessWindowFunction<Orders, Object, Integer, TimeWindow>.Context context, Iterable<Orders> elements, Collector<Object> collector) 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();collector.collect("key=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}});process.print();process.getSideOutput(lateData).print("侧输出流");environment.execute();}
}

这篇关于FlinkAPI开发之水位线（Watermark）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/613162。

相关文章

这15个Vue指令，让你的项目开发爽到爆

这15个Vue指令，让你的项目开发爽到爆

1. V-Hotkey 仓库地址: github.com/Dafrok/v-ho… Demo: 戳这里 https://dafrok.github.io/v-hotkey 安装: npm install --save v-hotkey 这个指令可以给组件绑定一个或多个快捷键。你想要通过按下 Escape 键后隐藏某个组件，按住 Control 和回车键再显示它吗？小菜一碟： <template
阅读更多...
水位雨量在线监测系统概述及应用介绍

水位雨量在线监测系统概述及应用介绍

在当今社会，随着科技的飞速发展，各种智能监测系统已成为保障公共安全、促进资源管理和环境保护的重要工具。其中，水位雨量在线监测系统作为自然灾害预警、水资源管理及水利工程运行的关键技术，其重要性不言而喻。 一、水位雨量在线监测系统的基本原理 水位雨量在线监测系统主要由数据采集单元、数据传输网络、数据处理中心及用户终端四大部分构成，形成了一个完整的闭环系统。 数据采集单元：这是系统的“眼睛”，
阅读更多...
Hadoop企业开发案例调优场景

Hadoop企业开发案例调优场景

需求 （1）需求：从1G数据中，统计每个单词出现次数。服务器3台，每台配置4G内存，4核CPU，4线程。 （2）需求分析： 1G / 128m = 8个MapTask；1个ReduceTask；1个mrAppMaster 平均每个节点运行10个 / 3台 ≈ 3个任务（4    3    3） HDFS参数调优 （1）修改：hadoop-env.sh export HDFS_NAMENOD
阅读更多...
嵌入式QT开发：构建高效智能的嵌入式系统

嵌入式QT开发：构建高效智能的嵌入式系统

摘要： 本文深入探讨了嵌入式 QT 相关的各个方面。从 QT 框架的基础架构和核心概念出发，详细阐述了其在嵌入式环境中的优势与特点。文中分析了嵌入式 QT 的开发环境搭建过程，包括交叉编译工具链的配置等关键步骤。进一步探讨了嵌入式 QT 的界面设计与开发，涵盖了从基本控件的使用到复杂界面布局的构建。同时也深入研究了信号与槽机制在嵌入式系统中的应用，以及嵌入式 QT 与硬件设备的交互，包括输入输出设
阅读更多...
OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息，即可快速实现设备配网，是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递，利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力，实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后，即可发送相关配网数据。同时还支持与常规Sof
阅读更多...
活用c4d官方开发文档查询代码

活用c4d官方开发文档查询代码

当你问AI助手比如豆包，如何用python禁止掉xpresso标签时候，它会提示到 这时候要用到两个东西。https://developers.maxon.net/论坛搜索和开发文档 比如这里我就在官方找到正确的id描述 然后我就把参数标签换过来
阅读更多...
Linux_kernel驱动开发11

Linux_kernel驱动开发11

一、改回nfs方式挂载根文件系统         在产品将要上线之前，需要制作不同类型格式的根文件系统         在产品研发阶段，我们还是需要使用nfs的方式挂载根文件系统         优点：可以直接在上位机中修改文件系统内容，延长EMMC的寿命         【1】重启上位机nfs服务         sudo service nfs-kernel-server resta
阅读更多...
【区块链 + 人才服务】区块链集成开发平台 | FISCO BCOS应用案例

【区块链 + 人才服务】区块链集成开发平台 | FISCO BCOS应用案例

随着区块链技术的快速发展，越来越多的企业开始将其应用于实际业务中。然而，区块链技术的专业性使得其集成开发成为一项挑战。针对此，广东中创智慧科技有限公司基于国产开源联盟链 FISCO BCOS 推出了区块链集成开发平台。该平台基于区块链技术，提供一套全面的区块链开发工具和开发环境，支持开发者快速开发和部署区块链应用。此外，该平台还可以提供一套全面的区块链开发教程和文档，帮助开发者快速上手区块链开发。
阅读更多...
Vue3项目开发——新闻发布管理系统（六）

Vue3项目开发——新闻发布管理系统（六）

文章目录 八、首页设计开发1、页面设计2、登录访问拦截实现3、用户基本信息显示①封装用户基本信息获取接口②用户基本信息存储③用户基本信息调用④用户基本信息动态渲染 4、退出功能实现①注册点击事件②添加退出功能③数据清理 5、代码下载 八、首页设计开发 登录成功后，系统就进入了首页。接下来，也就进行首页的开发了。 1、页面设计 系统页面主要分为三部分，左侧为系统的菜单栏，右侧
阅读更多...
v0.dev快速开发

v0.dev快速开发

探索v0.dev：次世代开发者之利器 今之技艺日新月异，开发者之工具亦随之进步不辍。v0.dev者，新兴之开发者利器也，迅速引起众多开发者之瞩目。本文将引汝探究v0.dev之基本功能与优势，助汝速速上手，提升开发之效率。 何谓v0.dev？ v0.dev者，现代化之开发者工具也，旨在简化并加速软件开发之过程。其集多种功能于一体，助开发者高效编写、测试及部署代码。无论汝为前端开发者、后端开发者
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2