flink
水位线
处理时间
事件时间
自定义水位线策略
用来生成水位线的方法:.assignTimestampsAndWatermarks(),他主要用来为流中的数据分配时间戳,并生成水位线来指示事件时间
stream.assignTimestampsAndWatermarks(WatermarkStrategy<T> watermarkStrategy)
//指定怎么从数据中指定时间戳,如果自定义Strategy可以不写
.withTimestampAssigner();
.assignTimestampsAndWatermarks()方法需要传入一个 WatermarkStrategy 作为参数,这就是 所 谓 的 “ 水 位 线 生 成 策 略 ” 。 WatermarkStrategy 中 包 含 了 一 个 “ 时 间 戳 分 配 器”TimestampAssigner 和一个“水位线生成器”WatermarkGenerator。
public interface WatermarkStrategy<T> extends TimestampAssignerSupplier<T>,WatermarkGeneratorSupplier<T>{
@Override
TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);
@Override
WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}
⚫ TimestampAssigner:主要负责从流中数据元素的某个字段中提取时间戳,并分配给元素。时间戳的分配是生成水位线的基础。
⚫ WatermarkGenerator:主要负责按照既定的方式,基于时间戳生成水位线。在WatermarkGenerator 接口中,主要又有两个方法:onEvent()和 onPeriodicEmit()。
⚫ onEvent:每个事件(数据)到来都会调用的方法,它的参数有当前事件、时间戳,以及允许发出水位线的一个 WatermarkOutput,可以基于事件做各种操作
⚫ onPeriodicEmit:周期性调用的方法,可以由 WatermarkOutput 发出水位线。周期时间为处理时间,可以调用环境配置的.setAutoWatermarkInterval()方法来设置,默认为200m
有序流使用WatermarkStrategy.forMonotonousTimestamps()即可
示例
public static class CustomPeriodicGenerator implements WatermarkGenerator<Event> {
private Long delayTime = 5000L; // 延迟时间
private Long maxTs = Long.MIN_VALUE + delayTime + 1L; // 观察到的最大时间戳
@Override
public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
// 每来一条数据就调用一次
maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
// 发射水位线,默认 200ms 调用一次 -1是因为时间窗口左闭右开
output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
}
}
聚合函数(AggregateFunction)
富函数 RichFuction
有生命周期,有open和close方法,可以获取运行时上下文环境。
标签:WatermarkStrategy,Flink,笔记,水位,时间,WatermarkOutput,public,WatermarkGenerator From: https://www.cnblogs.com/ljtstudy/p/16867407.html