下篇·flink基于rocksdb的timerService

发表于 | 分类于 | | |
字数统计 2.2k | 阅读时长 9

[toc]

接上文分析，要将timer改成基于rocksdb，其实就是要对存储timer的set和queue提供基于rocksdb的存储方案。以下我们基于flink1.7版本源码分析

registerProcessingTimeTimer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public void registerProcessingTimeTimer(N namespace, long time) {
	InternalTimer<K, N> oldHead = processingTimeTimersQueue.peek();
	if (processingTimeTimersQueue.add(new TimerHeapInternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace))) {
		long nextTriggerTime = oldHead != null ? oldHead.getTimestamp() : Long.MAX_VALUE;
		// check if we need to re-schedule our timer to earlier
		// 如果新加入的timer的时间更早触发，那么就需要把先前的timer取消
		if (time < nextTriggerTime) {
			if (nextTimer != null) {
				nextTimer.cancel(false);
			}
			nextTimer = processingTimeService.registerTimer(time, this);
		}
	}
}

可以和看到1.4版本中的基本逻辑是一致的，只是存储方式变化了，下面我们就来分析一下新的存储方式是怎么实现的。在存储的选择上依然有Heap和RocksDB两个方式

1
2
3
4
5
6
7
8
9
10
switch (priorityQueueStateType) {
			case HEAP:
				this.priorityQueueFactory = new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
				break;
			case ROCKSDB:
				this.priorityQueueFactory = new RocksDBPriorityQueueSetFactory();
				break;
			default:
				throw new IllegalArgumentException("Unknown priority queue state type: " + priorityQueueStateType);
		}

从timerService的需求来看我们可以看到这样的几个需求：

  1. 能够每次poll出最近需要触发的timer，实际上是需要维护一个小顶堆
  2. 能够对每一个key的timer去重

针对这两个需求，总体来看基于Heap的实现是通过基于数组实现了一个二叉堆，具体实现类为HeapPriorityQueue, 然后针对去重的功能又继承该PQ，通过一个hashmap数组，数组的每一个元素代表一个KG的一组不重复timer，同时这组timer内部也维护了timer在二叉堆中存储的下标，方便deleteTimer时的快速删除。

基于Heap的实现

HeapPriorityQueueElement,AbstractHeapPriorityQueue,HeapPriorityQueue,HeapPriorityQueueSet

  1. 存储基于数组，通过HeapPriorityQueueElement记录自己所在的index，可以达到快速删除的目的
  2. 数组中存储的是一个二叉树，数组的起始位置是从1开始，为了使一些热点方法做更少的计算

在实现上是采用template的设计模式，主要实现逻辑交由子类来实现:

  • addInternal
  • removeInternal
  • getHeadElementIndex
addInternal
1
2
3
4
public boolean add(@Nonnull T toAdd) {
	addInternal(toAdd);
	return toAdd.getInternalIndex() == getHeadElementIndex();
}

添加一个timer至数组中，返回值false表示队首的元素没有改变，true则表示改变了或者不确定

1
2
3
4
5
protected void addInternal(@Nonnull T element) {
	final int newSize = increaseSizeByOne();
	moveElementToIdx(element, newSize);
	siftUp(newSize);
}
1
2
3
4
5
6
7
8
9
10
11
private int increaseSizeByOne() {
	final int oldArraySize = queue.length;
	final int minRequiredNewSize = ++size;
	if (minRequiredNewSize >= oldArraySize) {
		final int grow = (oldArraySize < 64) ? oldArraySize + 2 : oldArraySize >> 1;
		// 当存储元素的个数大于数组长度时，需要进行扩容，通过`Arrays.copyOf`进行数组内容的拷贝
		resizeQueueArray(oldArraySize + grow, minRequiredNewSize);
	}
	// TODO implement shrinking as well?
	return minRequiredNewSize;
}
1
2
3
4
5
// 将新加入的元素存储到相应的idx处，并且记录该元素在queue中的位置
protected void moveElementToIdx(T element, int idx) {
		queue[idx] = element;
		element.setInternalIndex(idx);
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
private void siftUp(int idx) {
		final T[] heap = this.queue;
		final T currentElement = heap[idx];
		int parentIdx = idx >>> 1;
		
		// 每次将比较的index，缩小一半，如果被比较元素的优先级高于新插入的元素就将被比较元素后移，直至比较到第一个元素。这样能够保证idx为1的元素是最早时间触发的
		while (parentIdx > 0 && isElementPriorityLessThen(currentElement, heap[parentIdx])) {
			moveElementToIdx(heap[parentIdx], idx);
			idx = parentIdx;
			parentIdx >>>= 1;
		}

		moveElementToIdx(currentElement, idx);
	}
1
2
3
4
// 比较两个值的优先级
private boolean isElementPriorityLessThen(T a, T b) {
		return elementPriorityComparator.comparePriority(a, b) < 0;
	}
removeInternal
  1. 抽取第一个timer用以触发
  2. 用户删除某个timer的行为

删除的方式是通过idx下标来实现快速删除的，这也就是HeapPriorityQueueElement中记录idx的作用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
protected T removeInternal(int removeIdx) {
	T[] heap = this.queue;
	T removedValue = heap[removeIdx];

	// 要删除的idx应该和内部存储value值保存的idx一致
	assert removedValue.getInternalIndex() == removeIdx;

	final int oldSize = size;

	// 删除的不是数组的最后一个元素需要进行位置的调整
	if (removeIdx != oldSize) {
		T element = heap[oldSize];
		// 将原先的最后一个元素放置到要删除的idx处，但是这样的放置没有考虑优先级
		moveElementToIdx(element, removeIdx);
		adjustElementAtIndex(element, removeIdx);
	}

	heap[oldSize] = null;

	--size;
	return removedValue;
}
1
2
3
4
5
6
private void adjustElementAtIndex(T element, int index) {
	siftDown(index);
	if (queue[index] == element) {
		siftUp(index);
	}
}
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
private void siftDown(int idx) {
	final T[] heap = this.queue;
	final int heapSize = this.size;

	final T currentElement = heap[idx];
	int firstChildIdx = idx << 1;
	int secondChildIdx = firstChildIdx + 1;

	if (isElementIndexValid(secondChildIdx, heapSize) &&
		isElementPriorityLessThen(heap[secondChildIdx], heap[firstChildIdx])) {
		firstChildIdx = secondChildIdx;
	}

	while (isElementIndexValid(firstChildIdx, heapSize) &&
		isElementPriorityLessThen(heap[firstChildIdx], currentElement)) {
		moveElementToIdx(heap[firstChildIdx], idx);
		idx = firstChildIdx;
		firstChildIdx = idx << 1;
		secondChildIdx = firstChildIdx + 1;

		if (isElementIndexValid(secondChildIdx, heapSize) &&
			isElementPriorityLessThen(heap[secondChildIdx], heap[firstChildIdx])) {
			firstChildIdx = secondChildIdx;
		}
	}

	moveElementToIdx(currentElement, idx);
}

以上的操作大致上是一个二叉堆的增删的调整过程，涉及的具体算法可以查阅下文末的资料。

以下来分析rocksdb存储的实现

KeyGroupPartitionedPriorityQueue

基于rocksdb的存储是通过这个KeyGroupPartitionedPriorityQueue类来实现的，这个类中通过一个内存优先级队列，也就是上文中提到的内部实现的HeapPriorityQueue，用以存储所有KG的timer，而每一个分组的timer是如何存储呢？

1
2
3
4
5
6
for (int i = 0; i < keyGroupedHeaps.length; i++) {
			final PQ keyGroupSubHeap =
				orderedCacheFactory.create(firstKeyGroup + i, totalKeyGroups, keyExtractor, elementPriorityComparator);
			keyGroupedHeaps[i] = keyGroupSubHeap;
			heapOfKeyGroupedHeaps.add(keyGroupSubHeap);
		}

在这里的构造函数可以看到，其实是通过orderedCacheFactory,从字面意思看是一个有序的缓存，也就是为每一个KG创建一个有序的缓存类，并将其添加到优先级队列中，这里的subHeap也是一个可比较的类，相当于去取这两个subHeap的堆顶的元素拿出来比较下就可以知道这两个subheap的排序方式了。

比如poll的逻辑,首先先从HeapPQ中挑出堆顶（一个subPQ），然后再从这个PQ中取出堆顶就是要触发的timer了，而这个subPQ就是真是数据（timer）存储的地方了。

1
2
3
4
5
6
public T poll() {
	final PQ headList = heapOfKeyGroupedHeaps.peek();
	final T head = headList.poll();
	heapOfKeyGroupedHeaps.adjustModifiedElement(headList);
	return head;
}

RocksDBCachingPriorityQueueSet

这个是上节中subHeap的实现类

1
2
3
4
5
6
7
8
9
10
11
12
private void checkRefillCacheFromStore() {
		// 不是所有的元素都在cache(treeset)中，并且cache为空  
		if (!allElementsInCache && orderedCache.isEmpty()) {
			try (final RocksBytesIterator iterator = orderedBytesIterator()) {
				// 捞取rocksdb中这个columnFamily的部分数据填充treeset至maxsize
				orderedCache.bulkLoadFromOrderedIterator(iterator);
				allElementsInCache = !iterator.hasNext();
			} catch (Exception e) {
				throw new FlinkRuntimeException("Exception while refilling store from iterator.", e);
			}
		}
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public E peek() {

		checkRefillCacheFromStore();

		if (peekCache != null) {
			// 这个是维护的全局变量，只有在堆顶改变后在会置为null
			return peekCache;
		}

		byte[] firstBytes = orderedCache.peekFirst();
		if (firstBytes != null) {
			peekCache = deserializeElement(firstBytes);
			return peekCache;
		} else {
			return null;
		}
	}
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
public E poll() {

		checkRefillCacheFromStore();

		final byte[] firstBytes = orderedCache.pollFirst();

		if (firstBytes == null) {
			return null;
		}

		// write-through sync
		// 为什么不需要删除treeset中的元素呢？
		removeFromRocksDB(firstBytes);

		// 删除了这个columnFamily最后一个元素
		if (orderedCache.isEmpty()) {
			seekHint = firstBytes;
		}

		// 少一步反序列化的操作
		if (peekCache != null) {
			E fromCache = peekCache;
			peekCache = null;
			return fromCache;
		} else {
			return deserializeElement(firstBytes);
		}
	}
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
public boolean add(@Nonnull E toAdd) {

		checkRefillCacheFromStore();

		final byte[] toAddBytes = serializeElement(toAdd);

		final boolean cacheFull = orderedCache.isFull();

		// 如果cache没满并且之前所有元素都在cache中了  或者新加入的元素的优先级通过byte数组的优先级比较发现应该在堆顶
		if ((!cacheFull && allElementsInCache) ||
			LEXICOGRAPHIC_BYTE_COMPARATOR.compare(toAddBytes, orderedCache.peekLast()) < 0) {

			if (cacheFull) {
				// we drop the element with lowest priority from the cache
				orderedCache.pollLast();
				// the dropped element is now only in the store
				allElementsInCache = false;
			}

			// 用来判重
			if (orderedCache.add(toAddBytes)) {
				// write-through sync
				addToRocksDB(toAddBytes);
				if (toAddBytes == orderedCache.peekFirst()) {
					// 说明新的写入导致了堆顶变化
					peekCache = null;
					return true;
				}
			}
		} else {
			// 如果cache满了，或者不是所有的元素都在cache中，说明新来的数据一定不是堆顶的数据
			// we only added to the store
			addToRocksDB(toAddBytes);
			allElementsInCache = false;
		}
		return false;
	}
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
public boolean remove(@Nonnull E toRemove) {

		checkRefillCacheFromStore();

		final byte[] oldHead = orderedCache.peekFirst();

		if (oldHead == null) {
			return false;
		}

		final byte[] toRemoveBytes = serializeElement(toRemove);

		// write-through sync
		removeFromRocksDB(toRemoveBytes);
		orderedCache.remove(toRemoveBytes);

		if (orderedCache.isEmpty()) {
			seekHint = toRemoveBytes;
			peekCache = null;
			return true;
		}

		if (oldHead != orderedCache.peekFirst()) {
			peekCache = null;
			return true;
		}

		return false;
	}

Iterator中seekHint的作用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private RocksBytesIterator(@Nonnull RocksIteratorWrapper iterator) {
			this.iterator = iterator;
			try {
				// We use our knowledge about the lower bound to issue a seek that is as close to the first element in
				// the key-group as possible, i.e. we generate the next possible key after seekHint by appending one
				// zero-byte.
				iterator.seek(Arrays.copyOf(seekHint, seekHint.length + 1));
				currentElement = nextElementIfAvailable();
			} catch (Exception ex) {
				// ensure resource cleanup also in the face of (runtime) exceptions in the constructor.
				iterator.close();
				throw new FlinkRuntimeException("Could not initialize ordered iterator.", ex);
			}
		}

再说checkpoint

在doc中作者也提到之所以要做这个feature除了因为timer过多会导致OOM等问题，还有一个原因是因为timer的属性虽然和keyed state很类似，但是代码管理以及checkpoint的方式都是单独的一块逻辑，并且checkpoint的持久化过程还是同步的（因为是以raw keyedstate的方式去进行的），再修改之后，每次注册的timeservice都会注册到kvstatInfo中，将checkpoint的逻辑统一到statebackend中并且实现了异步化。

https://blog.csdn.net/u010224394/article/details/8834969

https://github.com/apache/flink/pull/6159

https://docs.google.com/document/d/1XbhJRbig5c5Ftd77d0mKND1bePyTC26Pz04EvxdA7Jc/edit

谢谢支持