flink-watermark

发表于 | 分类于 | | |
字数统计 3.1k | 阅读时长 12

关于Flink中watermark工作原理代码分析

前言

waterMark，latency，checkpoint这三者实现方式都是上游节点逐步广播消息给下游节点来处理的行为（都是在流中插入一种特殊的数据结构来做处理）

时间语义

谈及watermark就要先从Flink支持的时间语义说起，Flink支持三种时间语义：

1
2
3
process time:指的系统处理对应数据时的系统时间。他是最简单的一种实现，由于不需要额外的协调，因性能最好
event time:是指数据中携带的时间，而不是数据到达的时间。因此时间的进度完全取决于数据，而不是系统时间。使用event time必须指定生成eventTime和watermark的方式。因为他一般会等待迟到的数据，因此一定会有一定的延时
ingestion time:是指数据进入flink的时间，在source处插入的时间，和process time一样无法处乱序事件

对于eventtime和ingestion time两种语义到达的数据有可能乱序的。从事件产生（例如日志采集数据中的乱序日志），到流经source，再到operator，中间是有一个过程时间的。虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、背压等原因，导致乱序的产生（out-of-order或者说late element）。
但是对于late element，我们又不能无限期的等下去，必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了。这个特别的机制，就是watermark，它告诉了算子时间不大于 WaterMark 的消息不应该再被接收【如果出现意味着延迟到达】。也就是说水位线以下的数据均已经到达。WaterMark 从源算子开始 emit，并逐级向下游算子传递。当源算子关闭时，会发射一个携带 Long.MAX_VALUE 值时间戳的 WaterMark，下游算子接收到之后便知道不会再有消息到达。

waterMark产生方式

waterMark的产生:有两种方式来产生watermark和timestamp

  1. 在数据源处直接进行assign timestamp 和generate watermark
  2. 通过timestamp和watermark generate operator来产生，如果使用了timestamp assigner和watermark generator在source处产生的timestamp和watermark会被覆盖。
  3. 其实这两种最终的实现方式还是一样的！即第一种也是在source处分配了timestampAssigner

方法一 在数据源处产生发送

目前实现了在source提取时间和产生operator的只有0.10版本的kafka fetcher中实现
首先根据应用StreamExecutionEnvironment#setStreamTimeCharacteristic设置的时间语义（存储于StreamConfig类）来获取
对应的sourceContext，sourceContext接口是sourceFunction发送数据的抽象，有三个实现类，根据时间语义划分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
switch (timeCharacteristic) {
            case EventTime:
                ctx = new ManualWatermarkContext<>(checkpointLock, output);
                break;
            case IngestionTime:
                ctx = new AutomaticWatermarkContext<>(processingTimeService, checkpointLock, output, watermarkInterval);
                break;
            case ProcessingTime:
                ctx = new NonTimestampContext<>(checkpointLock, output);
                break;
            default:
                throw new IllegalArgumentException(String.valueOf(timeCharacteristic));
        }
        return ctx;

这里我们就考虑eventTime的情况。
在kafka AbstractFetcher中还提供了三种模式来控制自己生产时间戳和watermark

1
2
3
NO_TIMESTAMPS_WATERMARKS: fetcher 不生产 timestamp 和 watermarks
PERIODIC_WATERMARKS: fetcher 阶段性定时生产 watermarks
PUNCTUATED_WATERMARKS: fetcher 生产标记 watermark 【按照特定的消息字段值触发】

这里的区分方式来自于FlinkKafkaConsumerBase#assignTimestampsAndWatermarks分配的AssignerWithPunctuatedWatermarks或AssignerWithPeriodicWatermarks 也就是assigner决定了产生方式，这其实就是把方法二给封装到了KafkaConsumerBase里面！当然这个也需要用户在创建consumer之后自定义一个assigner。

方法二 使用TimestampAssigner来实现

TimestampAssinger是接收一个流，并产生一个新的流带上了时间戳和watermark，如果原来的流已经带有了timestamp和watermark那么这个将会被覆盖。timestamp assinger只需要在关于时间的操作之前加上即可。

flink通过接口TimestampAssigner来让用户依据消息的格式自己抽取可能被用于 WaterMark的timestamp，它只定义了一个接口：

1
long extractTimestamp(T element, long previousElementTimestamp);

previousElementTimestamp这个参数传入的是什么呢？

1
2
3
final long newTimestamp = userFunction.extractTimestamp(value, 
				element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE);
因此如果一个元素如果已经带有timestamp，比如在source处已经分配，那么在这里处理的时候会被覆盖掉。

而TimestampAssigner的两个继承接口AssignerWithPunctuatedWatermarks以及 AssignerWithPeriodicWatermarks 定义了waterMark生成的两种典型方式。

AssignerWithPeriodicWatermarks

AssignerWithPeriodicWatermarks是周期性的产生watermark，每过一定间隔 ExecutionConfig#getAutoWatermarkInterval()，系统会调用getCurrentWatermark来获取最新的waterMark值，如果新的waterMark值有增长那么就会发送一个新的waterMark，如果没有新的元素进来，那么getCurrentWatermark则不会被周期性的调用，这个接口的好处是，可以定义最大乱序时间，减少因为数据延迟到达而被时间窗口丢弃的行为，实现类BoundedOutOfOrdernessTimestampExtractor

AssignerWithPunctuatedWatermarks

AssignerWithPunctuatedWatermarks的使用场景是针对在接收到特定的elements之后才触发更新waterMark的操作。比如有一个流中有一些元素带有flag表示没有晚于这个元素时间的元素了,那么他的实现代码是这样的

1
2
3
4
5
6
7
8
9
10
public class WatermarkOnFlagAssigner implements AssignerWithPunctuatedWatermarks<MyElement> {

    public long extractTimestamp(MyElement element, long previousElementTimestamp) {
         return element.getSequenceTimestamp();
     }

     public Watermark checkAndGetNextWatermark(MyElement lastElement, long extractedTimestamp) {
         return lastElement.isEndOfSequence() ? new Watermark(extractedTimestamp) : null;
     }
 }
小结

我们平常用的最多的还是AssignerWithPeriodicWatermarks 并设置数据到达最大超时时长。下面的分析我们就以AssignerWithPeriodicWatermarks为例。
想到一个点AssignerWithPunctuatedWatermarks的一个使用场景:可能比较适合于数据不是连续发送或者说是批任务的场景，比如说是每天某时候数据有更新之后才有计算，那么只要在进入的最后一个数据打入endFlag，然后进行waterMark更新触发数据处理（一个想法，尚未实践）。

waterMark在传输及对window Operator的作用方式

TimestampsAndPeriodicWatermarksOperator

在DataStream调用assignTimestampsAndWatermarks产生了一个TimestampsAndPeriodicWatermarksOperator

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
	
public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(
	AssignerWithPeriodicWatermarks<T> timestampAndWatermarkAssigner) {
		
	// match parallelism to input, otherwise dop=1 sources could lead to some strange
	// behaviour: the watermark will creep along very slowly because the elements
	// from the source go to each extraction operator round robin.
	//这里就是说一般会默认将并发度设成和inputOperator的并发度一致，避免因为elements进入extraction operator的时候要随机进入分区。
	final int inputParallelism = getTransformation().getParallelism();
	final AssignerWithPeriodicWatermarks<T> cleanedAssigner = clean(timestampAndWatermarkAssigner);
		
	TimestampsAndPeriodicWatermarksOperator<T> operator = 
			new TimestampsAndPeriodicWatermarksOperator<>(cleanedAssigner);
		
	return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator)
			.setParallelism(inputParallelism);
}

我们看这个TimestampsAndPeriodicWatermarksOperator

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@Override
public void open() throws Exception {
	super.open();

	currentWatermark = Long.MIN_VALUE;
	watermarkInterval = getExecutionConfig().getAutoWatermarkInterval();
	
	if (watermarkInterval > 0) {
		long now = getProcessingTimeService().getCurrentProcessingTime();
		getProcessingTimeService().registerTimer(now + watermarkInterval, this);
	}
}

	@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定期watermarkInterval发送userFunction.getCurrentWatermark()用户定义的waterMark,当然也是要waterMark有上涨的情况下才会发送

而且这里有个有意思的地方,这里也定义了processWatermark方法，该方法主要调用时机主要，是在StreamInputProcessor#processInput中,这个类在后面再具体分析

1
2
3
4
5
6
7
8
public void processWatermark(Watermark mark) throws Exception {
	// if we receive a Long.MAX_VALUE watermark we forward it since it is used
	// to signal the end of input and to not block watermark progress downstream
	if (mark.getTimestamp() == Long.MAX_VALUE && currentWatermark != Long.MAX_VALUE) {
		currentWatermark = Long.MAX_VALUE;
		output.emitWatermark(mark);
	}
}

但是这里覆盖了AbstractStreamOperator中的写法是为什么呢?

1
2
3
4
5
6
public void processWatermark(Watermark mark) throws Exception {
	for (HeapInternalTimerService<?, ?> service : timerServices.values()) {
		service.advanceWatermark(mark.getTimestamp());
	}
	output.emitWatermark(mark);
}

这里其实体现了在最后一个TimestampsAndPeriodicWatermarksOperator之前定义的waterMark传递此operator的时候除非是一个流结束标志Long.MAX_VALUE,否则不会发送，只会有该operator定时发送waterMark给下游处理，这也就说明了在source处定义的waterMark会被后面定义的给覆盖。

所以大部分Operator处理waterMark的方式是AbstractStreamOperator中定义好的

1
2
3
4
5
6
public void processWatermark(Watermark mark) throws Exception {
	for (HeapInternalTimerService<?, ?> service : timerServices.values()) {
		service.advanceWatermark(mark.getTimestamp());//这个处理逻辑还要在看下
	}
	output.emitWatermark(mark);
}

这里看到他是会给所有的下游channel发送一个watermark

1
2
3
4
5
public void emitWatermark(Watermark mark) {
	for (Output<StreamRecord<OUT>> out : allOutputs) {
		out.emitWatermark(mark);
	}
}

StreamInputProcessor

processWatermark是在StreamInputprocessor中调用的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
if (recordOrMark.isWatermark()) {
		long watermarkMillis = recordOrMark.asWatermark().getTimestamp();
		if (watermarkMillis > watermarks[currentChannel]) {
			watermarks[currentChannel] = watermarkMillis;
			long newMinWatermark = Long.MAX_VALUE;
			for (long watermark: watermarks) {
				newMinWatermark = Math.min(watermark, newMinWatermark);
			}
			if (newMinWatermark > lastEmittedWatermark) {
				lastEmittedWatermark = newMinWatermark;
				synchronized (lock) {
					streamOperator.processWatermark(new Watermark(lastEmittedWatermark));
				}
			}
		}
		continue;
	}

逻辑总结:（channel的理解还不够）

  1. 如果消费到的消息是一个 WaterMark，获得其对应的 source channel id 并将时间更新进去，同时记录下当前所有 channel 的最小 WaterMark 时间
  2. 如果当前最小 WaterMark 时间【所有的 channel 都至少消费到该时间】大于上次发射给下游的 WaterMark 时间，则更新 WaterMark 时间并将其交给算子处理
  3. 通常算子在处理【尤其是涉及了窗口计算或者需要时间缓存策略的算子】后会将 WaterMark 继续往下游广播发送

waterMark对window作用形式

waterMark如何触发窗口计算

情况一: late element

  • Event Time < watermark时间（对于late element太多的数据而言）,这种情况下只要来一条数据就会触发窗口计算，其他属于该窗口的数据到达后都会被丢弃。

情况二：乱序

  • watermark时间 >= window_end_time（对于out-of-order以及正常的数据而言）
  • 在[window_start_time,window_end_time)中有数据存在

window的触发机制，是先按照自然时间将window划分，如果window大小是3秒，那么1分钟内会把window划分为如下的形式:

1
2
3
4
[00:00:00,00:00:03)
[00:00:03,00:00:06)
...
[00:00:57,00:01:00)

如果window大小是10秒，则window会被分为如下的形式：当然还有一个offset值可以控制window的起始值不是整点。

1
2
3
4
[00:00:00,00:00:10)
[00:00:10,00:00:20)
...
[00:00:50,00:01:00)

1
2
3
4
5
6
7
8
9
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中当ctx.getCurrentWatermark > window.maxTimestamp时立刻触发窗口计算.

当然没有内容是不会触发计算的:

1
2
3
4
5
6
7
if (triggerResult.isFire()) {
					ACC contents = windowState.get();
					if (contents == null) {
						continue;
					}
					emitWindowContents(actualWindow, contents);
				}

输入的数据中，根据自身的Event Time，将数据划分到不同的window中，如果window中有数据，则当watermark时间>=window_edn_time时，就符合了window触发的条件了，最终决定window触发，还是由数据本身的Event Time所属的window中的window_end_time决定。

以上代码和情况二相符，其实情况一也是情况二的特殊情况，watermark > 数据的 eventtime也就是说超过了最大的延迟时间，此时其实也是来了之后watermark > window.endtime，然后必然会被触发只是其他该窗格的数据会被丢弃罢了。

数据清理的逻辑

1
2
3
protected boolean isLate(W window) {
	return (windowAssigner.isEventTime() && (cleanupTime(window) <= internalTimerService.currentWatermark()));
}
1
2
3
4
5
6
7
8
private long cleanupTime(W window) {
		if (windowAssigner.isEventTime()) {
			long cleanupTime = window.maxTimestamp() + allowedLateness;
			return cleanupTime >= window.maxTimestamp() ? cleanupTime : Long.MAX_VALUE;
		} else {
			return window.maxTimestamp();
		}
	}

小结

最大乱序时间要结合自己的业务以及数据情况去设置。如果maxOutOfOrderness设置的太小，而自身数据发送时由于网络等原因导致乱序或者late太多，那么最终的结果就是会有很多单条的数据在window中被触发，数据的正确性影响太大，对此可以通过在windowOperator处添加metrics监控来指导业务方设置成什么样的一个值才是最合理的。

allowLatency

最后一点上文代码里提到的这个allowLatency又有什么作用呢?

默认情况下当watermark涨过了window的endtime之后，再有属于该窗口的数据到来的时候该数据会被丢弃，设置了allowLatency这个值之后，也就是定义了数据在watermark涨过window.endtime但是又在allowlatency之前到达的话仍旧会被加到对应的窗口去。会使得窗口再次被触发。Flink会保存窗口的状态直到allow latenness 超期。

待补充：

  1. 对流传输过程中channel的理解
  2. keyby操作过程对watermark的影响
  3. advanceWatermark操作理解

参考

http://blog.csdn.net/lmalds/article/details/52704170
http://chenyuzhao.me/2017/02/09/flink-watermark-checkpoint/
https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/event_timestamps_watermarks.html
https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/windows.html#allowed-lateness

谢谢支持