Flink水位线不触发的几点总结

窗口计算时遇到好几次水位线不触发的情况,简单总结下。

首先,介绍下Flink的事件时间(EventTime)和水位线(Watermarks)的概念。

一、处理时间

如果要构造一个实时的流式应用,或早或晚都会接触到EventTime这个概念。现实场景中也会遇到消息乱序到达,这里会介绍到为什么需要事件时间和如何去处理乱序到达的数据。
ProcessingTime是Flink系统处理这条消息的时间,EventTime可以理解成是这条消息真实发生的时间。
举个例子,创建一个SlidingWindow,窗口大小为10秒,步长为5秒。关于窗口的更多概念,可以参考Flink官方文档——Windows

案例1:消息都按时到达

1
2
3
4
5
6
7
val text = senv.socketTextStream("localhost", 9999)
val counts = text.map {(m: String) => (m.split(",")(0), 1) }
.keyBy(0)
.timeWindow(Time.seconds(10), Time.seconds(5))
.sum(1)
counts.print
senv.execute("ProcessingTime processing example")

如果source中有三条消息,对应的事件时间分别为13秒、13秒和16秒:
pr_ino_source
它们会落到正确的窗口上,如下图所示。13秒产生的两条消息会落到窗口1[5s-15s]和窗口2[10s-20s]上,16秒产生的消息会落到窗口2[10s-20s]和窗口3[15s-25s]上。最后窗口fire掉时,三个窗口的count值分别为:(a,2), (a,3) and (a,1) ,和预期一致。
pr_ino_windows

案例2:消息delay

其中一条13秒产生的消息晚到了6s,那按上面的代码逻辑,这些消息会落到下面的窗口中:
pr_ooo_windows
延迟的消息会落到窗口2[10s-20s]和窗口3[15s-25s]上。这看起来对窗口2没有影响,因为结果都是3,但窗口3的结果却不一致了。

二、事件时间

因此此处我们采用事件时间,这里水位线的事件时间为当前系统的时间,当然你可以改成数据中的某个时间。

1
2
3
4
5
6
7
8
class TimestampExtractor extends AssignerWithPeriodicWatermarks[String] with Serializable {
override def extractTimestamp(e: String, prevElementTimestamp: Long) = {
e.split(",")(1).toLong
}
override def getCurrentWatermark(): Watermark = {
new Watermark(System.currentTimeMillis)
}
}

1
2
3
4
5
6
7
8
9
senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val text = senv.socketTextStream("localhost", 9999)
.assignTimestampsAndWatermarks(new TimestampExtractor)
val counts = text.map {(m: String) => (m.split(",")(0), 1) }
.keyBy(0)
.timeWindow(Time.seconds(10), Time.seconds(5))
.sum(1)
counts.print
senv.execute("EventTime processing example")

结果如下图所示:
ev_ooo_windows
结果看起来好很多,窗口2和3都正确了,但窗口1却丢了一条数据。
Flink没有将delay的数据分配给窗口3是因为现在是检查消息的事件时间,因此不会放入窗口三中。而没有分配给窗口1的原因是delayed的消息到达系统的时间是19秒,窗口1已经fire掉了。此处就需要watermarks了。

三、水位线(水印)

我认为水位线是很重要和有趣的一个概念,我这里会大概描述下,如果想了解更多可以看Google一场精彩的talk,也可以看这个dataArtisans的blog。水位线简单理解就是一个timestamp,当Flink收到这个水印时,Flink理解会收到来自这个时间点之后的消息,也可以理解成告诉Flink运行到哪个事件时间了。
在这个案例中,其实就是告诉Flink一条消息可以迟到多久。
我现在设置水位线为现在的事件提前5秒,相当于告诉Flink我的消息可以迟到五秒。

1
2
3
override def getCurrentWatermark(): Watermark = { 
new Watermark(System.currentTimeMillis - 5000)
}

结果变成下图所示:
ev_ooo_windows_wat

四、允许延迟(Lateness)

如果采用“watermark - delay”,如果水位线不超过window_length + delay是不会被fire掉的,所以此刻可以采用allowedLateness方法。在window_end_time + allowed lateness之前,Flink都不会丢弃这条数据。
当消息到达时,Flink会提取它的时间,然后判断它是否在有效的延迟时间内,然后去判断是否fire掉窗口。
但是通过这种途径,一个窗口有可能被fire掉多次,如果需要exactly once processing的话,需要保证sink是幂等的。

五、水位线怎么不触发?

数据一直有序得进来,为什么没有窗口被fire掉?没有结果产出?

case1:提取时间失败

笔者和上游约定好了数据格式,extractTimestamp中提取的是某个字段为事件时间。研究数据发现约定好的字段突然不发了。

case2:提取时间有了,但是照样失败

上游按约定发了该字段后,系统在测试环境运行了一段时间,又没有结果产出了。
调试发现提取时间正常,checkAndGetNextWatermark也正常,但是为什么窗口没被fire掉呢。
因为时间的format变了,由到毫秒的timestamp变成了yyyymmddHHmmss,需要转成timestamp。

case3:一切正常,窗口不fire

1
2
3
4
5
6
7
8
9
10
11
EventTimeTrigger.java
@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// if the watermark is already past the window fire immediately
return TriggerResult.FIRE;
} else {
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}

看了下EventTimeTrigger的源码,只有当当前的水位线越过窗口,即时间大于窗口的endTime才会触发Fire的操作。我们的处理流程没有触发,那就说明我们的水位线没有更新到合适的值。调试后发现当前的水位线一直停留在初始的最小的long值。

1
2
3
4
5
6
7
8
9
10
BoundedOutOfOrdernessTimestampExtractor.java
@Override
public final Watermark getCurrentWatermark() {
// this guarantees that the watermark never goes backwards.
long potentialWM = currentMaxTimestamp - maxOutOfOrderness;
if (potentialWM >= lastEmittedWatermark) {
lastEmittedWatermark = potentialWM;
}
return new Watermark(lastEmittedWatermark);
}

debug发现lastEmittedWatermark确实有更新,这说明这个地方是触发了Watermark的值。但是debug的过程中,发现时不时会出现初始值的水位线。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
SystemProcessingTimeService.java
TimestampsAndPeriodicWatermarksOperator.java
@Override
public void onProcessingTime(long timestamp) throws Exception {
// register next timer
Watermark newWatermark = userFunction.getCurrentWatermark();
if (newWatermark != null && newWatermark.getTimestamp() > currentWatermark) {
currentWatermark = newWatermark.getTimestamp();
// emit watermark
output.emitWatermark(newWatermark);
}
long now = getProcessingTimeService().getCurrentProcessingTime();
getProcessingTimeService().registerTimer(now + watermarkInterval, this);
}

TimestampsAndPeriodicWatermarksOperator会做判断:如果新的水位线小于当前的水位线,就不会更新了。
终于,顺着StreamInputProcessor–>StatusWatermarkValve理了下来,看见这样的处理逻辑:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
StreamInputProcessor.java
StatusWatermarkValve.java
private void findAndOutputNewMinWatermarkAcrossAlignedChannels() {
long newMinWatermark = Long.MAX_VALUE;
// determine new overall watermark by considering only watermark-aligned channels across all channels
for (InputChannelStatus channelStatus : channelStatuses) {
if (channelStatus.isWatermarkAligned) {
newMinWatermark = Math.min(channelStatus.watermark, newMinWatermark);
}
}
// we acknowledge and output the new overall watermark if it is larger than the last output watermark
if (newMinWatermark > lastOutputWatermark) {
lastOutputWatermark = newMinWatermark;
outputHandler.handleWatermark(new Watermark(lastOutputWatermark));
}
}

这里会将所有的channel status的水位线做个汇总:取最小的水位线。那是不是问题出在这里?后面debug了下看看,确实,这个地方有的channel status下的水位线一直是最小的long值那个不正常的水位线,进而导致整体的水位线发送不出去。
那么为什么会出现这种情况呢,百思不得其解。
当 [window_start_time,window_end_time) 有数据,watermark Time大于等于window_end_time时,会触发window trigger。
因为之前运行都是正常的,检查了数据也没问题。去翻改动,有影响的可能就是改了一些算子的并行度。
assigntimestampandwatermarks和map的并行度一样了就不能生成水位线了?
于是修改了assigntimestampandwatermarks的并行度,window正常fire掉了。

分析原因:
Flink source用到了FlinkKafkaConsumer010,没有指定KafkaPartitioner的话,会通过FixedPartitioner来给出默认的partitioner方法:

1
2
3
4
public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
Preconditions.checkArgument(partitions != null && partitions.length > 0, "Partitions of the target topic is empty.");
return partitions[this.parallelInstanceId % partitions.length];
}

parallelInstanceId代表着Flink consumer程序的并行度ID,假如FlinkKafkaConsumer010的并行度是12,那么这12个线程的ID分别是1-12.
parallelInstances代表着总的并行度,即12。
partitions是一个kafka partition的数组,kafka的topic的partitions是4(因为性能测试,换到只有一个节点的kafka)。
Flink partition的规则,就是Flink的并行度ID除以kafka partition length取余。
因此kafka编号为1-4的partition分别对应source node的1-4的partition,那么source node5-12的partition就为空了。
默认的partition策略是按照Flink的并行度ID与kafka中partition的数量取余的方法分配的,而与数据本身没有关系。
source node的partition为5-12的接收不到数据,
Watermark的生成是数据驱动的,只有当TimestampAndWatermarkAssigner”看到”数据时,watermark才会生效。
而map和assigntimestampandwatermarks并发度一样的话,这八个partition的watermark不会修改,所以会出现watermark的初始值一直存在的情况。
当assigntimestampandwatermarks的并行度修改后,事件会被shuffled across(洗牌),因此到了TimestampAndWatermarkAssigner不会有空的partition存在了。
以上。