首先，介绍下Flink的事件时间(EventTime)和水位线(Watermarks)的概念。
如果要构造一个实时的流式应用，或早或晚都会接触到EventTime这个概念。现实场景中也会遇到消息乱序到达，这里会介绍到为什么需要事件时间和如何去处理乱序到达的数据。
ProcessingTime是Flink系统处理这条消息的时间，EventTime可以理解成是这条消息真实发生的时间。
举个例子，创建一个SlidingWindow，窗口大小为10秒，步长为5秒。关于窗口的更多概念，可以参考Flink官方文档——Windows。
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) ，和预期一致。
其中一条13秒产生的消息晚到了6s，那按上面的代码逻辑，这些消息会落到下面的窗口中：
延迟的消息会落到窗口2[10s-20s]和窗口3[15s-25s]上。这看起来对窗口2没有影响，因为结果都是3，但窗口3的结果却不一致了。
因此此处我们采用事件时间，这里水位线的事件时间为当前系统的时间，当然你可以改成数据中的某个时间。
![[站外图片上传中...(image-1e2304-1550219427900)]](http://pangongsen.com/uploads/watermarks/pr_ino_source.png){kind=link}
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我的消息可以迟到五秒。
![[站外图片上传中...(image-b2feb1-1550219427900)]](http://pangongsen.com/uploads/watermarks/ev_ooo_windows.png){kind=link}
override def getCurrentWatermark(): Watermark = {
new Watermark(System.currentTimeMillis - 5000)
}
结果变成下图所示:
[站外图片上传中...(image-6dfb6e-1550219427900)]
如果采用“watermark - delay”，如果水位线不超过window_length + delay是不会被fire掉的，所以此刻可以采用allowedLateness方法。在window_end_time + allowed lateness之前，Flink都不会丢弃这条数据。
当消息到达时，Flink会提取它的时间，然后判断它是否在有效的延迟时间内，然后去判断是否fire掉窗口。
但是通过这种途径，一个窗口有可能被fire掉多次，如果需要exactly once processing的话，需要保证sink是幂等的。
数据一直有序得进来，为什么没有窗口被fire掉？没有结果产出？
笔者和上游约定好了数据格式，extractTimestamp中提取的是某个字段为事件时间。研究数据发现约定好的字段突然不发了。
上游按约定发了该字段后，系统在测试环境运行了一段时间，又没有结果产出了。
调试发现提取时间正常，checkAndGetNextWatermark也正常，但是为什么窗口没被fire掉呢。
因为时间的format变了，由到毫秒的timestamp变成了yyyymmddHHmmss，需要转成timestamp。
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存在了。
以上。
