watermark 如何下发的?
一、 数据读取
A. AbstractStreamTaskNetworkInput：该类是用于读取上游数据
//processElement
private void processElement(StreamElement recordOrMark, DataOutput<T> output) throws Exception {
if (recordOrMark.isRecord()) { //数据
output.emitRecord(recordOrMark.asRecord());
} else if (recordOrMark.isWatermark()) { // watermark
statusWatermarkValve.inputWatermark(
recordOrMark.asWatermark(), flattenedChannelIndices.get(lastChannel), output);
} else if (recordOrMark.isLatencyMarker()) { // latencymarker
output.emitLatencyMarker(recordOrMark.asLatencyMarker());
} else if (recordOrMark.isStreamStatus()) { //stream status
statusWatermarkValve.inputStreamStatus(
recordOrMark.asStreamStatus(),
flattenedChannelIndices.get(lastChannel),
output);
} else {
throw new UnsupportedOperationException("Unknown type of StreamElement");
}
}
- 对象类别 含义：
StreamStatus ： idle(空闲) ， active(非空闲)
watermark ： 水位
- 什么是LatencyMarker?
https://blog.csdn.net/nazeniwaresakini/article/details/106615777
二、watermark下发逻辑
watermark下发是上游读取数据，判断每个channel watermark进行对齐并下发新watermark的过程。
StatusWatermarkValve：
- 概念：
a. InputChannelStatus[] channelStatuses: // 每个channel的状态
protected long watermark; // channel的watermark
protected StreamStatus streamStatus; // channel当前的状态
protected boolean isWatermarkAligned; // watermark是否对齐
b. private long lastOutputWatermark; // 上次发出的watermark
c. private StreamStatus lastOutputStreamStatus; // 上次输出的streamstatus
isWatermarkAligned 不对齐的情况:
(1)streamStatus 为idle
(2) streamstatus是active但是watermark还没有达到上次watermark
- 初始化
每个channelstatus 的watermark取的最小值，streamstatus是active状态，watermarkaligned是对齐的。lastoutputwatermark是最小值，lastoutputstreamstatus也是对齐的。
public StatusWatermarkValve(int numInputChannels) {
checkArgument(numInputChannels > 0);
this.channelStatuses = new InputChannelStatus[numInputChannels];
for (int i = 0; i < numInputChannels; i++) {
channelStatuses[i] = new InputChannelStatus();
channelStatuses[i].watermark = Long.MIN_VALUE; //初始化watermark
channelStatuses[i].streamStatus = StreamStatus.ACTIVE; //初始化streamstatus
channelStatuses[i].isWatermarkAligned = true; //初始化watermark是对齐的
}
this.lastOutputWatermark = Long.MIN_VALUE;
this.lastOutputStreamStatus = StreamStatus.ACTIVE;
}
image.png
灵魂提问:
- 为什么要引入watermark?
Apache Flink uses watermarks to keep track of the progress in event time. The event time is extracted from one of the fields of the data event that contain the timestamp when that event was originally created.- 为什么要设置idleness?
- 事件偏差
（1） Flink 作业需要使用来自具有不同特征的源的事件
随着时间的推移，使用这种源组合将导致它们各自的水印进程彼此显着不同。这通常在 Flink 作业失败后被放大，作业需要赶上读取事件。
（2）数据本身的分布不平衡。例如，某些键可能比其他键更频繁地出现，这使得正在处理这些键的相应运算符只需要做更多的处理，这可能会导致水印进程变慢。- 造成的问题
因为watermark变小，导致待定计时器的数量将增加，与之相关的资源量也会增加，背压在增加，检查点屏障的进度将减慢，整个检查点过程将花费更长的时间，导致输入流被阻塞，进而进一步减慢事件的处理速度。这些副作用最终可能导致内存不足错误、检查点失败甚至作业崩溃。- 解决办法：
从 Flink 1.11 开始，对所谓的空闲检测提供了开箱即用的支持。如果一个主题在较长时间内没有产生事件并因此导致没有水印进展，则该主题被称为空闲。Flink 提供的解决方案是，当在可配置的时间段内没有收到任何事件时，将此输入源标记为“暂时空闲”，从而在确定算子的最低水印时忽略此源。
如前所述，在读取一组具有不同特征的输入源时，需要应用不同的解决方案。我们为此提出的，就是我们所说的“平衡阅读”。平衡读取既可以在单个 Kafka 消费者层面实现，也可以在 Flink 作业中所有 Kafka 消费者的各种实例之间实现。
疑问：https://stackoverflow.com/questions/48576391/flink-how-set-up-initial-watermark
源码解析
1. inputwatermark
public void inputWatermark(Watermark watermark, int channelIndex, DataOutput<?> output)
throws Exception {
// TODO 为啥忽略 当前channel 或者 lastOutputStreamStatus 是idle的情况
if (lastOutputStreamStatus.isActive()
&& channelStatuses[channelIndex].streamStatus.isActive()) {
long watermarkMillis = watermark.getTimestamp();
//当前watermark > 当前channel的watermark， 小于的情况可以不处理,为了channel上watermark是升序的对齐的
if (watermarkMillis > channelStatuses[channelIndex].watermark) {
channelStatuses[channelIndex].watermark = watermarkMillis;
// 如果watermark赶上上次watermark，那么不对齐的channels会被设为对齐
if (!channelStatuses[channelIndex].isWatermarkAligned
&& watermarkMillis >= lastOutputWatermark) {
channelStatuses[channelIndex].isWatermarkAligned = true;
}
// 尝试找到所有对齐的channel里的最小的watermark
findAndOutputNewMinWatermarkAcrossAlignedChannels(output);
}
}
}
// 对齐的channel里，取最小值 作为下发的watermark 并且大于上次watermark 才输出
private void findAndOutputNewMinWatermarkAcrossAlignedChannels(DataOutput<?> output)
throws Exception {
long newMinWatermark = Long.MAX_VALUE;
boolean hasAlignedChannels = false;
// 新的watermark 由 所有 watermark对齐的channel 产生，取最小值
for (InputChannelStatus channelStatus : channelStatuses) {
if (channelStatus.isWatermarkAligned) { //当前channel watermark对齐，取最小值
hasAlignedChannels = true;
newMinWatermark = Math.min(channelStatus.watermark, newMinWatermark);
}
}
// 当有channel事对齐的，且当前watermark最小值大于上次watermark最小值，下发当前watermark
if (hasAlignedChannels && newMinWatermark > lastOutputWatermark) {
lastOutputWatermark = newMinWatermark;
output.emitWatermark(new Watermark(lastOutputWatermark));
}
}
结论： watermark的处理
（1）需要保证在每个channel上 watermark是升序的才处理
（2）下发watermark 是通过判断对齐的channel（没有标记空闲的)的最小值需要大于lastoutputwatermark。也就是说下发的watermark也要是升序的。
2. inputStreamStatus
public void inputStreamStatus(StreamStatus streamStatus, int channelIndex, DataOutput<?> output)
throws Exception {
// 当前streamstaus空闲，但改channel之前是active的，设置该channel空闲且不对齐
if (streamStatus.isIdle() && channelStatuses[channelIndex].streamStatus.isActive()) {
//设置状态从 active -> idle
channelStatuses[channelIndex].streamStatus = StreamStatus.IDLE;
// 设置对不对齐
channelStatuses[channelIndex].isWatermarkAligned = false;
// 当这个channel是最后一个空闲的，也就是当前所有的都空闲，我们需要输出streamstatus为idle
if (!InputChannelStatus.hasActiveChannels(channelStatuses)) {
// now that all input channels are idle and no channels will continue to advance its
// watermark,we should "flush" all watermarks across all channels; effectively, this means
// emitting the max watermark across all channels as the new watermark. Also, since we
// already try to advance the min watermark as channels individually become IDLE, here we only need to
// perform the flush if the watermark of the last active channel that just became idle is the current
// min watermark.
// 因为当前所有channel空闲，并且channel都无法继续更新水印，应该flush所有channel上的watermark，
// 意味着我们设置channel 中最大的watermark作为新水印
// 因为在其他变为idle的时候仍然会下发min watermark，所以只需要在最后一个channel变成idle且等于上次min watermark的时候flush
if (channelStatuses[channelIndex].watermark == lastOutputWatermark) {
findAndOutputMaxWatermarkAcrossAllChannels(output);
}
//设置为空闲，并下发整个流是空闲的状态
lastOutputStreamStatus = StreamStatus.IDLE;
output.emitStreamStatus(lastOutputStreamStatus);
//有非空闲的时候，并且当前watermark等于上次watermark,取最小的watermark
} else if (channelStatuses[channelIndex].watermark == lastOutputWatermark) {
// if the watermark of the channel that just became idle equals the last output
// watermark (the previous overall min watermark), we may be able to find a new
// min watermark from the remaining aligned channels
// 如果watemark不是最后一个变为空闲，并且等于上次outputwatermark，我们只需要从剩余的对齐的channel取最小值作为watermark
findAndOutputNewMinWatermarkAcrossAlignedChannels(output);
}
// streamstatus非空闲 并且之前是空闲
} else if (streamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isIdle()) {
// 设置channel的状态从idle变为active
channelStatuses[channelIndex].streamStatus = StreamStatus.ACTIVE;
// 上次watermark大于等于上次output watermark,则认为channel的 watermark是对齐的
if (channelStatuses[channelIndex].watermark >= lastOutputWatermark) {
channelStatuses[channelIndex].isWatermarkAligned = true;
}
// 如果上次整体streamstatus标记为空闲，改为active并输出最后状态
if (lastOutputStreamStatus.isIdle()) {
lastOutputStreamStatus = StreamStatus.ACTIVE;
output.emitStreamStatus(lastOutputStreamStatus);
}
}
}
// 取所有channel里的最大值
private void findAndOutputMaxWatermarkAcrossAllChannels(DataOutput<?> output) throws Exception {
long maxWatermark = Long.MIN_VALUE;
for (InputChannelStatus channelStatus : channelStatuses) {
maxWatermark = Math.max(channelStatus.watermark, maxWatermark);
}
if (maxWatermark > lastOutputWatermark) {
lastOutputWatermark = maxWatermark;
output.emitWatermark(new Watermark(lastOutputWatermark));
}
}
解释：
channel从active变为idle：
(1) 如果这是最后一个（active->idle）的channel:
a.那么flush整个流中的watermark。方法是 只有在channel的watermark等于上次outputwatermark,取所有channel最大值作为outputchannel下发。
b.下游传递streamstatus为空闲状态。
(2) 不是最后一个（active->idle）的channel，但是当前channel等于上次outputwatermark,取非空闲的channel的最小值。channel从idle变为active:
a. 当前channel的watermark大于等于lastoutputwatermark,是指该channel为对齐
b.上次输出的streamstatus是idle,输出streamstatus改为active
