1.正文
前面，我们已经学过了 一文搞懂 Flink 处理 Barrier 全过程，今天我们一起来看一下 flink 是如何处理水印的，以 Flink 消费 kafka 为例
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(topics, new SimpleStringSchema(), properties);
consumer.setStartFromLatest();
consumer.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
@Override
public long extractAscendingTimestamp(String element) {
String locTime = "";
try {
Map<String, Object> map = Json2Others.json2map(element);
locTime = map.get("locTime").toString();
} catch (IOException e) {
}
LocalDateTime startDateTime =
LocalDateTime.parse(locTime, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
long milli = startDateTime.toInstant(OffsetDateTime.now().getOffset()).toEpochMilli();
return milli;
}
});
通过 assignTimestampsAndWatermarks 来对 watermarksPeriodic 进行赋值，当 KafkaFetcher ( 关于 KafkaFetcher 可以参考 写给大忙人看的Flink 消费 Kafka) 在初始化的时候，会创建 PeriodicWatermarkEmitter
// if we have periodic watermarks, kick off the interval scheduler
// 在构建 fetcher 的时候创建 PeriodicWatermarkEmitter 并启动，以周期性发送
if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
@SuppressWarnings("unchecked")
PeriodicWatermarkEmitter periodicEmitter = new PeriodicWatermarkEmitter(
subscribedPartitionStates,
sourceContext,
processingTimeProvider,
autoWatermarkInterval);
periodicEmitter.start();
}
PeriodicWatermarkEmitter 主要的作用就是周期性的发送 watermark，默认周期是 200 ms，通过 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 指定。
@Override
//每隔 interval 时间会调用此方法
public void onProcessingTime(long timestamp) throws Exception {
long minAcrossAll = Long.MAX_VALUE;
boolean isEffectiveMinAggregation = false;
for (KafkaTopicPartitionState<?> state : allPartitions) {
// we access the current watermark for the periodic assigners under the state
// lock, to prevent concurrent modification to any internal variables
final long curr;
//noinspection SynchronizationOnLocalVariableOrMethodParameter
synchronized (state) {
curr = ((KafkaTopicPartitionStateWithPeriodicWatermarks<?, ?>) state).getCurrentWatermarkTimestamp();
}
minAcrossAll = Math.min(minAcrossAll, curr);
isEffectiveMinAggregation = true;
}
// emit next watermark, if there is one
// 每隔 interval 对 watermark 进行合并取其最小的 watermark 发送至下游算子，并且保持单调递增
if (isEffectiveMinAggregation && minAcrossAll > lastWatermarkTimestamp) {
lastWatermarkTimestamp = minAcrossAll;
emitter.emitWatermark(new Watermark(minAcrossAll));// StreamSourceContexts.ManualWatermarkContext,watermark 与 record 的发送路径是分开的
}
// schedule the next watermark
timerService.registerTimer(timerService.getCurrentProcessingTime() + interval, this);
}
其中 PeriodicWatermarkEmitter 最关键性的方法就是 onProcessingTime。做了两件事
- 在保持水印单调性的同时合并各个 partition 的水印( 即取各个 partition 水印的最小值 )
- 注册 process timer 以便周期性的调用 onProcessingTime
接下来就是进行一系列的发送，与 StreamRecord 的发送过程类似，具体可以参考 一文搞定 Flink 消费消息的全流程
下游算子通过 StreamInputProcessor.processInput 方法接受到 watermark 并处理
......
// 如果元素是 watermark，就准备更新当前 channel 的 watermark 值（并不是简单赋值，因为有乱序存在）
if (recordOrMark.isWatermark()) {
// handle watermark
statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
continue;
// 如果元素是 status，就进行相应处理。可以看作是一个 flag，标志着当前 stream 接下来即将没有元素输入（idle），
// 或者当前即将由空闲状态转为有元素状态（active）。同时，StreamStatus 还对如何处理 watermark 有影响。
// 通过发送 status，上游的 operator 可以很方便的通知下游当前的数据流的状态。
} else if (recordOrMark.isStreamStatus()) {
// handle stream status 把对应的 channelStatuse 改为 空闲，
// 然后如果所有的 channelStatuse 都是 idle 则找到最大的 watermark 并处理，否则找到最小的 watermark 并处理
statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);
continue;
}
......
进入 StatusWatermarkValve.inputWatermark watermark 真正被处理的地方
//当水印输入时的处理操作
public void inputWatermark(Watermark watermark, int channelIndex) {
// ignore the input watermark if its input channel, or all input channels are idle (i.e. overall the valve is idle).
// streamStatus 和 channelStatus 都是 active 才继续往下计算
if (lastOutputStreamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isActive()) {
long watermarkMillis = watermark.getTimestamp();
// if the input watermark's value is less than the last received watermark for its input channel, ignore it also.
// ignore 小于已经接收到的 watermark 的 watermark，保持其单调性
if (watermarkMillis > channelStatuses[channelIndex].watermark) {
channelStatuses[channelIndex].watermark = watermarkMillis;
// previously unaligned input channels are now aligned if its watermark has caught up
// 如果之前未对齐的，现在对齐。
if (!channelStatuses[channelIndex].isWatermarkAligned && watermarkMillis >= lastOutputWatermark) {
channelStatuses[channelIndex].isWatermarkAligned = true;
}
// ow, attempt to find a new min watermark across all aligned channels
findAndOutputNewMinWatermarkAcrossAlignedChannels();
}
}
}
isWatermarkAligned 其实就是由于之前是 idle，无需关心 watermark， 现在有数据了，需要关心 watermark 了。
而 findAndOutputNewMinWatermarkAcrossAlignedChannels 其实就是取 所有 channel 中的最小值，并且在保证 watermark 单调递增的情况下处理 watermark, 调用了 StreamInputProcessor.handleWatermark
@Override
public void handleWatermark(Watermark watermark) {
try {
synchronized (lock) {
watermarkGauge.setCurrentWatermark(watermark.getTimestamp());//gauge
//处理 watermark 的入口
operator.processWatermark(watermark);
}
} catch (Exception e) {
throw new RuntimeException("Exception occurred while processing valve output watermark: ", e);
}
}
我们以 AbstractStreamOperator 为例看具体是如何处理 watermark 的
public void processWatermark(Watermark mark) throws Exception {//operator.processWatermark(mark)
if (timeServiceManager != null) {//有 timeService 则不为 null 如 window InternalTimeServiceManager
//timeService
timeServiceManager.advanceWatermark(mark);
}
//处理结束之后继续往下游发送依次循环。
output.emitWatermark(mark);
}
当 filter、flatMap 等算子 timeServiceManager 均等于 null，我们以 windowOperator 为例，看 timeServiceManager.advanceWatermark(mark); 如何操作的
public void advanceWatermark(Watermark watermark) throws Exception {
for (InternalTimerServiceImpl<?, ?> service : timerServices.values()) {
service.advanceWatermark(watermark.getTimestamp());//处理 watermark event time 对于 trigger 的调用逻辑
}
}
继续调用
public void advanceWatermark(long time) throws Exception {//watermark timestamp
currentWatermark = time;
InternalTimer<K, N> timer;
while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) {//自定义，触发一系列满足条件的 key
eventTimeTimersQueue.poll();
keyContext.setCurrentKey(timer.getKey());//
// 触发 triggerTarget.onEventTime
triggerTarget.onEventTime(timer);
}
}
当注册的 eventTimer 的 timestamp <= currentwatermark 时，触发 triggerTarget.onEventTime(timer); 即调用 WindowOperator.onEventTime 方法
关于 windowOperator 的具体细节可以参考 写给大忙人看的 Flink Window原理
// 这个是通过 timer 来调用的
// window processElement 的时候 registerCleanupTimer(window) window.maxTimestamp()+allowedLateness
// 和 eventTrigger onElement registerEventTimeTimer(window.maxTimestamp()) 会创建相应的 timer
// 所以当 allowedLateness 不为 0 的时候，同一个 window.maxTimestamp() 对应的 eventWindow 会触发两次,
// 而且默认 windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp()) 才会清除 window state
public void onEventTime(InternalTimer<K, W> timer) throws Exception {
triggerContext.key = timer.getKey();
triggerContext.window = timer.getNamespace();
MergingWindowSet<W> mergingWindows;
if (windowAssigner instanceof MergingWindowAssigner) {
mergingWindows = getMergingWindowSet();
W stateWindow = mergingWindows.getStateWindow(triggerContext.window);
if (stateWindow == null) {
// Timer firing for non-existent window, this can only happen if a
// trigger did not clean up timers. We have already cleared the merging
// window and therefore the Trigger state, however, so nothing to do.
return;
} else {
windowState.setCurrentNamespace(stateWindow);
}
} else {
windowState.setCurrentNamespace(triggerContext.window);
mergingWindows = null;
}
TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
if (triggerResult.isFire()) {
ACC contents = windowState.get();
if (contents != null) {
emitWindowContents(triggerContext.window, contents);
}
}
if (triggerResult.isPurge()) {
windowState.clear();
}
if (windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
clearAllState(triggerContext.window, windowState, mergingWindows);
}
if (mergingWindows != null) {
// need to make sure to update the merging state in state
mergingWindows.persist();
}
}
关于窗口的触发有三种情况( 对应的源码部分可以参考 写给大忙人看的 Flink Window原理 )
- 然后就是当 time == window.maxTimestamp() 立即触发窗口
- window.maxTimestamp() <= ctx.getCurrentWatermark() 立即触发，即允许迟到范围内的数据到来
- window.maxTimestamp() + allowedLateness<= ctx.getCurrentWatermark() ,主要是为了针对延迟数据，保证数据的准确性
2.总结
水印的处理其实还蛮简单的，分两部分
1. 水印在满足单调递增的情况下，要么直接发往下游( OneInputStreamOperator,像 keyby、filter、flatMap ),
要么取最小值然后发往下游( TwoInputStreamOperator,像 co系列 coFlatMap、IntervalJoinOperator、TemporalJoin)
2. 设置水印时间为当前 StreamRecord 中的时间戳，此时间戳是<= watermark ，因为 watermark 是单调递增的，而 StreamRecord 的时间戳就是提取出来的时间戳
