Flink 学习五 Flink 时间语义
1.时间语义
在流式计算中.时间是一个影响计算结果非常重要的因素! (窗口函数,定时器等)
Flink 可以根据不同的时间概念处理数据。
- 处理时间: process time
System.currentTimeMillis()是指执行相应操作的机器系统时间（也称为纪元时间，例如 Java 的时间）。是现实世界的时间,时间是单调递增的 - 事件时间: event time 是指根据附加到每一行的时间戳处理流式数据。时间戳可以对事件发生的时间进行编码。是业务数据中的时间,时间有可能停滞,但是不会回溯,时间是不可会退的;
2.时间语义API设置
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //事件时间
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // 处理时间
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime); // 算子到达时间 其实就是处理时间
上述代码在1.12 版本之前, 已TimeCharacteristic.ProcessingTime 作为默认的时间语义,也可以用上述代码设置时间语义;
1.12 版本之后,flink 以 TimeCharacteristic.EventTime 作为时间语义 ,并且Deprecated上面的代码 在使用需要指定时间语义的API 时,在显示的指定对应的时间语义;
keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(5),Time.seconds(2)));
keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(2)));
keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)));
keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(2)));
也可以禁用 event time 语义
/**
* Sets the interval of the automatic watermark emission. Watermarks are used throughout the
* streaming system to keep track of the progress of time. They are used, for example, for time
* based windowing.
*
* <p>Setting an interval of {@code 0} will disable periodic watermark emission.
*
* @param interval The interval between watermarks in milliseconds.
*/
@PublicEvolving
public ExecutionConfig setAutoWatermarkInterval(long interval) {
Preconditions.checkArgument(interval >= 0, "Auto watermark interval must not be negative.");
this.autoWatermarkInterval = interval;
return this;
}
3.watermark 简介
上面说到1.12 版本之后,flink 以 TimeCharacteristic.EventTime 作为时间语义,Flink 收到数据中的事件时间有可能不是有序的,这就导致会收到迟到的数据,其事件时间是属于过去的窗口;
为了能够基于事件时间进行计算,Flink引入了Watermark的概念
- watermark ,本质上也是flink各个算子之间流转的一种标记数据,flink内部自动产生并插入到数据流里面的
- 消息流信息中就是时间戳
watermark 的产生源头:
- watermark 一般是 来源于source 算子
- source 算子计算出来的watermark 广播到下面
4.watermark 状态
4.1.初始watermark
一般是来源于source 算子
在watermark 产生的源头算子中,subTask 程序会用一个定时器,去周期性的检查收到的数据的时间的最大值,如果超过了之前记录的最大值,就把这个最大值更新为watermark,并下游算子广播(通过API设置数据中那个字段作为事件时间)
4.2.下游/中间算子的watermark
中间算子收到上游算子广播的watermark ,其算子内部也会有一个定时器去定时的检测收到的所有的上游算子的watermark ,并计算其中最小值作为当前算子的watermark,并下游算子广播
注:当其中一个所有算子,不在更新watermark 怎么处理? flink 提供一个机制设置watermark的idletime,意思就是如果在idletime时间内没有收到上游算子广播的watermark,则会自动的往前面推进watermark
5.watermark生成策略
5.1 生成watermark时机
Flink 1.12 版本之后,watermark 的生产策略是固定频率周期性的产生
- AssignerWithPeriodicWatermarks 周期性生成
- AssignerWithPunctuatedWatermarks 指定标志生成,比如数据中的某个属性
5.1 生成watermark的数值
新版API watermark生成策略
- 紧跟最大事件时间(watermark=周期内最大时间) :WatermarkStrategy.forMonotonousTimestamps();
- 允许乱序(watermark=周期内最大时间-容错时间) :WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5));
- 自定义生成策略:WatermarkStrategy.forGenerator()
- 不生产watermark 禁用了时间推进:WatermarkStrategy.noWatermarks()
6.watermark 示例
简单的测试并发度为1下的算子watermark更新情况
public class _01_Watermark {
public static void main(String[] args) throws Exception {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// 获取环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> dataStreamSource = env.socketTextStream("192.168.141.141", 9000);
WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy.<String>forMonotonousTimestamps( )
.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
@Override
public long extractTimestamp(String element, long recordTimestamp) {
String s = element.split(",")[0];
long time = 0;
try {
time = simpleDateFormat.parse(s).getTime(); //解析字符串值转long 作为时间戳
} catch (ParseException e) {
}
return time;
}
});
SingleOutputStreamOperator<String> streamOperator = dataStreamSource.assignTimestampsAndWatermarks(watermarkStrategy);
SingleOutputStreamOperator<String> processed = streamOperator.process(new ProcessFunction<String, String>() {
@Override
public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
long currentWatermark = ctx.timerService().currentWatermark();
long processingTime = ctx.timerService().currentProcessingTime();
System.out.println("currentWatermark:" + simpleDateFormat.format(new Date(currentWatermark)));
System.out.println("currentWatermark:" + currentWatermark); //打印watermark
System.out.println("processingTime:" + simpleDateFormat.format(new Date(processingTime)));
out.collect(value);
}
});
processed.print();
env.execute ();
}
}
7.浅析watermark 源码
还是以上面示例为例讲解
7.1 准备
主要是 WatermarkStrategy.forMonotonousTimestamps( )和 WatermarkStrategy.forBoundedOutOfOrderness
两者对应的类是都是一样的 均为BoundedOutOfOrdernessWatermarks,只是 WatermarkStrategy.forMonotonousTimestamps( )对应BoundedOutOfOrdernessWatermarks中的参数 outOfOrdernessMillis = 0,后续主要是看BoundedOutOfOrdernessWatermarks类即可
@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {
//最大时间戳
private long maxTimestamp;
//可以延迟的毫秒值
private final long outOfOrdernessMillis;
/**设置初始水位
*/
public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");
this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
}
// ------------------------------------------------------------------------
//触发水位的更新
@Override
public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
}
//周期的触发,
@Override
public void onPeriodicEmit(WatermarkOutput output) {
output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
}
}
