Flink水位线不触发问题

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

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

一、处理时间

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

案例1：消息都按时到达

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秒：
[站外图片上传中...(image-1e2304-1550219427900)]
它们会落到正确的窗口上，如下图所示。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的结果却不一致了。

二、事件时间

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

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)
  }
}

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")

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

三、水位线（水印）

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

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

结果变成下图所示:

[站外图片上传中...(image-6dfb6e-1550219427900)]

四、允许延迟(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

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值。

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的过程中，发现时不时会出现初始值的水位线。

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理了下来，看见这样的处理逻辑:

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方法：

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存在了。

以上。

五 实际定位

1. 问题：发现程序没有输出，没有任何报错。
2. 可以先去任务的Appmaster上看watermark运行情况.来定位是否是watermarke的问题

image-20181219103115949

3. 如果是显示no watermark ，按照上述的case 进行诊断。
4. 在assignTimestampsAndWatermarks 后setParallelism(2) 。具体并行度根据实际需求设置，在这里kafkatopic只有2个分区，并行度设置为2