上篇·flink基于rocksdb的timerService

发表于 | 分类于 | | |
字数统计 1.6k | 阅读时长 7

[toc]

本文主要介绍flink中TimerService based on Rocksdb实现以及和之前版本的一个比较。

动机

  1. timer和keyed state分开单独管理，keyed state是由KeyedStateBackend管理，而timer是由InternalTimerServeice管理
  2. InternalTimerServeice现有的实现是基于heap的HeapInternalTimerService，当timer数量较多时会有OOM的问题.如果能像keyed state一样基于KeyedStateBackend管理，就能在timer数量比较多的时候选用rocksdb作为backend来解决扩展性的问题
  3. timer目前的checkpoint过程是通过raw keyed state的方式，在同步的过程中完成写出到外置存储，并且对于snapshot和restore timer都单独维护了一份代码。这块代码和其他的keyed state的实现有很多相同之处（分隔到keygroup来实现rescale，元数据的序列化和持久化..）

实现目标

  1. Have an implementation of timer services that operates on RocksDB.
  2. Support asynchronous snapshots for all timer state.
  3. Support incremental snapshots for timer state in RocksDB.
  4. Integrate timer state as another form of keyed state in keyed state backends in a way that leverages the existing snapshotting code to eliminate special casing code paths that do similar things. As as nice side effect, this would also free the raw keyed state for user state.

源码分析

首先我们来看一下1.4版本中timerService是怎么实现的,timerService 实现了以下两个接口：

1
2
3
4
5
6
7
8
9
10
11
InternalTimerService
    long currentProcessingTime();
    long currentWatermark();
    void registerProcessingTimeTimer(N namespace, long time);
    void deleteProcessingTimeTimer(N namespace, long time);
    void registerEventTimeTimer(N namespace, long time);
    void deleteEventTimeTimer(N namespace, long time);
提供的是当前时间获取和注册timer的方法

ProcessingTimeCallback
    void onProcessingTime(long timestamp) throws Exception;

在HeapInternalTimerService中分别维护了processing time和event time的timer集合

1
2
3
4
5
private final Set<InternalTimer<K, N>>[] processingTimeTimersByKeyGroup;
private final PriorityQueue<InternalTimer<K, N>> processingTimeTimersQueue;

private final Set<InternalTimer<K, N>>[] eventTimeTimersByKeyGroup;
private final PriorityQueue<InternalTimer<K, N>> eventTimeTimersQueue;

InternalTimer是一个Comparable, 按照timer触发的时间进行比较，timer是由key,namespace,timestamp唯一确定，其实也可以理解成如果同一个key，namespace下只可以有一个时间事件被触发。regsterTimer实际上就是在executor中注册一个任务

1
2
return timerService.schedule(
                    new TriggerTask(status, task, checkpointLock, target, timestamp), delay, TimeUnit.MILLISECONDS);

同时每个timerService中还维护了一个ProcessingTimeService用以处理和processing time相关的时间操作，是对ScheduledThreadPoolExecutor的包装，提供一些周期性执行和将来某时执行一次的操作.

我们在回头看HeapInternalTimerService的实现：

registerProcessingTimeTimer

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public void registerProcessingTimeTimer(N namespace, long time) {
    InternalTimer<K, N> timer = new InternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace);

    // make sure we only put one timer per key into the queue
    // 在存储timer的时候会存储两份，一份是通过Set的形式将各个Keygroup的区分开，另一份是按照时间顺序排除存储在一个`PriorityQueue`中
    Set<InternalTimer<K, N>> timerSet = getProcessingTimeTimerSetForTimer(timer);
    if (timerSet.add(timer)) {

        InternalTimer<K, N> oldHead = processingTimeTimersQueue.peek();
        long nextTriggerTime = oldHead != null ? oldHead.getTimestamp() : Long.MAX_VALUE;

        processingTimeTimersQueue.add(timer);

        // check if we need to re-schedule our timer to earlier
        // 如果新注册的timer比最近要触发的timer时间早，那么就会终止最近要触发的timer（如果已经跑起来了就不中断了）
        if (time < nextTriggerTime) {
            if (nextTimer != null) {
                nextTimer.cancel(false);
            }
            // 通过ScheduledThreadPoolExecutor注册一个task
            nextTimer = processingTimeService.registerTimer(time, this);
        }
    }
}

// 获取这个timer的key所属的已经注册的timer列表，从上面的注释我们可以看出是为了保证不注册重复timer
private Set<InternalTimer<K, N>> getProcessingTimeTimerSetForTimer(InternalTimer<K, N> timer) {
    checkArgument(localKeyGroupRange != null, "The operator has not been initialized.");
    int keyGroupIdx = KeyGroupRangeAssignment.assignToKeyGroup(timer.getKey(), this.totalKeyGroups);
    return getProcessingTimeTimerSetForKeyGroup(keyGroupIdx);
}

private Set<InternalTimer<K, N>> getProcessingTimeTimerSetForKeyGroup(int keyGroupIdx) {
    int localIdx = getIndexForKeyGroup(keyGroupIdx);
    Set<InternalTimer<K, N>> timers = processingTimeTimersByKeyGroup[localIdx];
    // 如过这个set没有出现过，就构建一个新的set存放这个key的timer
    if (timers == null) {
        timers = new HashSet<>();
        processingTimeTimersByKeyGroup[localIdx] = timers;
    }
    return timers;
}

private int getIndexForKeyGroup(int keyGroupIdx) {
    checkArgument(localKeyGroupRange.contains(keyGroupIdx),
        "Key Group " + keyGroupIdx + " does not belong to the local range.");
    return keyGroupIdx - this.localKeyGroupRangeStartIdx;
}

registerEventTimeTimer

1
2
3
4
5
6
7
public void registerEventTimeTimer(N namespace, long time) {
    InternalTimer<K, N> timer = new InternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace);
    Set<InternalTimer<K, N>> timerSet = getEventTimeTimerSetForTimer(timer);
    if (timerSet.add(timer)) {
        eventTimeTimersQueue.add(timer);
    }
}

可以看到eventtimer注册就不需要校验是否有将要执行的任务，因为eventtimer的实现不依赖于schedulerExxecutor。

onProcessingTime

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
27
28
29
30
// 这个方法是配合registerProcessingTimer，通过SystemProcessingTimeService来实现timer语义，在timer中注册的任务会回调这个onProcessing方法
public void onProcessingTime(long time) throws Exception {
    // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
    // inside the callback.
    // 如果不置为null，在执行triggerTarget.onProcessingTime(timer);里面执行了registerProcessingTimeTimer会调用本任务的cancel
    nextTimer = null;

    InternalTimer<K, N> timer;

    // 将processingTimeTimersQueue中所有小于当前时间的任务都取出进行出发
    while ((timer = processingTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) {

        // 删除set中存储的timer
        Set<InternalTimer<K, N>> timerSet = getProcessingTimeTimerSetForTimer(timer);

        timerSet.remove(timer);
        processingTimeTimersQueue.remove();

        // 每次触发之前需要设置当前的key
        keyContext.setCurrentKey(timer.getKey());
        triggerTarget.onProcessingTime(timer);
    }

    // 说明队列中还存在还没到时间需要触发的timer，需要注册新的FutureTask
    if (timer != null) {
        if (nextTimer == null) {
            nextTimer = processingTimeService.registerTimer(timer.getTimestamp(), this);
        }
    }
}

advanceWatermark

processing timer是基于executor来实现的，eventtime 的timer触发就依赖于watermark来触发，每次收到上游的watermark会触发调用advanceWatermark来将eventtime queue中的timer取出进行触发

1
2
3
4
5
6
public void processWatermark(Watermark mark) throws Exception {
    if (timeServiceManager != null) {
        timeServiceManager.advanceWatermark(mark);
    }
    output.emitWatermark(mark);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 这里的time就是最近的这次watermark的时间
public void advanceWatermark(long time) throws Exception {
    currentWatermark = time;

    InternalTimer<K, N> timer;

    // 同样是取出所有的小于watermark的timer进行触发
    while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) {

        Set<InternalTimer<K, N>> timerSet = getEventTimeTimerSetForTimer(timer);
        timerSet.remove(timer);
        eventTimeTimersQueue.remove();

        keyContext.setCurrentKey(timer.getKey());
        triggerTarget.onEventTime(timer);
    }
}

snapshot

之前在分析state实现的时候也分析过，在对operator进行snapshot的时候有一步就是对timerservice的数据进行snapshot

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
KeyGroupsList allKeyGroups = out.getKeyGroupList();
for (int keyGroupIdx : allKeyGroups) {
    out.startNewKeyGroup(keyGroupIdx);

    timeServiceManager.snapshotStateForKeyGroup(
        new DataOutputViewStreamWrapper(out), keyGroupIdx);
}


public void snapshotStateForKeyGroup(DataOutputView stream, int keyGroupIdx) throws IOException {
    InternalTimerServiceSerializationProxy<K, N> serializationProxy =
        new InternalTimerServiceSerializationProxy<>(timerServices, keyGroupIdx);

    serializationProxy.write(stream);
}

proxy这块主要是为兼容做了很多的工作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public void write(DataOutputView out) throws IOException {
    super.write(out);

    out.writeInt(timerServices.size());
    for (Map.Entry<String, HeapInternalTimerService<K, N>> entry : timerServices.entrySet()) {
        String serviceName = entry.getKey();
        HeapInternalTimerService<K, N> timerService = entry.getValue();

        out.writeUTF(serviceName);
        InternalTimersSnapshotReaderWriters
            .getWriterForVersion(VERSION, timerService.snapshotTimersForKeyGroup(keyGroupIdx))
            .writeTimersSnapshot(out);
    }
}

restore

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
protected void read(DataInputView in, boolean wasVersioned) throws IOException {
    int noOfTimerServices = in.readInt();

    for (int i = 0; i < noOfTimerServices; i++) {
        String serviceName = in.readUTF();

        HeapInternalTimerService<K, N> timerService = timerServices.get(serviceName);
        if (timerService == null) {
            timerService = new HeapInternalTimerService<>(
                totalKeyGroups,
                localKeyGroupRange,
                keyContext,
                processingTimeService);
            timerServices.put(serviceName, timerService);
        }

        int readerVersion = wasVersioned ? getReadVersion() : InternalTimersSnapshotReaderWriters.NO_VERSION;
        InternalTimersSnapshot<?, ?> restoredTimersSnapshot = InternalTimersSnapshotReaderWriters
            .getReaderForVersion(readerVersion, userCodeClassLoader)
            .readTimersSnapshot(in);

        timerService.restoreTimersForKeyGroup(restoredTimersSnapshot, keyGroupIdx);
    }

本文主要是对1.4版本的分析，下一篇文章基于1.7版本再分析timerservice on rocksdb的实现

参考:

https://docs.google.com/document/d/1XbhJRbig5c5Ftd77d0mKND1bePyTC26Pz04EvxdA7Jc/edit#heading=h.17v0k3363r6q

https://issues.apache.org/jira/browse/FLINK-9485

谢谢支持