文章/答案/技术大牛

发布

社区首页 >专栏 >Spillable StateBackend 之 SpillAndLoadManager 源码注解

Spillable StateBackend 之 SpillAndLoadManager 源码注解

原创

KyleMeow

修改于 2021-09-29 12:46:36

8870

文章被收录于专栏：腾讯云流计算 Oceanus腾讯云流计算 Oceanus

背景概要

在前文中，我们介绍了 Spillable Backend 及其 HeapStatusMonitor 的工作原理和不足。今天我们来看一下 Spillable Backend 的另一个核心组件：SpillAndLoadManager。如果说 HeapStatusMonitor 是测量系统负载的信号灯，那么 SpillAndLoadManager 就是具体的决策者，根据信号来决定当前系统的行为（将状态从内存 Spill 到磁盘，还是从磁盘 Load 到内存）。

SpillAndLoadManager 接口

SpillAndLoadManager 接口的定义非常简单，只有一个 checkResource() 方法，表示检查当前的资源，然后决定下一步的行动，其代码如下：

public interface SpillAndLoadManager {

	/**
	 * Check resource to decide whether to spill or load state.
	 */
	void checkResource();
}

目前的代码里，只有 SpillAndLoadManagerImpl 这一个实现类，因此本文以这个类为基准，通过逐行注解关键方法的方式，介绍它的工作原理，并对当前的实现方式提出一些建议。

SpillAndLoadManagerImpl 实现类初探

这个类目前有接近 500 行代码，处于中低复杂度。在分析它的工作原理之前，我们来首先先看一下他的整体结构：

其中上图的 Action 是一个枚举类型，有 SPILL、LOAD、NONE 三种取值，分别表示三种动作（NONE 表示”无动作“这种动作），因此很容易猜到，它是本方法的决策信号，决定了相关状态的去向。

ActionResult 则是对 Action 及比例参数（spillOrLoadRatio）的封装类，作为行动的指南：例如传入了 SPILL Action 和 0.2 的 spillOrLoadRatio 作为 ActionResult, 表明需要把 20% 的冷状态写入磁盘。

既然它的父类只定义了 checkResource() 一个方法，那我们先从这个方法入手，逐步分析该类的运行原理（下面的代码片段，删减了一些日志代码）。

public void checkResource() {
    long currentTime = System.currentTimeMillis();  // 获取当前 Unix 时间戳

    // 如果距离上次资源检查的时间小于阈值, 就不做检查. 配置项为 state.backend.spillable.resource-check.interval, 默认值为 10s
    if (currentTime - lastResourceCheckTime < resourceCheckInterval) { return; }

    lastResourceCheckTime = currentTime;    // 重置上次资源检查时间为当前值
    // getMonitorResult 访问的正是之前介绍的 heapStatusMonitor 里保存的资源负载。由于变量定义为了 volatile, 不能缓存, 因而访问开销较大
    HeapStatusMonitor.MonitorResult monitorResult = heapStatusMonitor.getMonitorResult();

    // 如果和上次获取的 monitorResult 相同, 则不做处理, 直接返回
    if (lastMonitorResult != null && lastMonitorResult.getId() == monitorResult.getId()) { return; }
    lastMonitorResult = monitorResult;

    ActionResult checkResult = decideAction(monitorResult);   // 根据 monitorResult 的值, 决定下一步的动作是 SPILL, LOAD 还是 NONE
    if (checkResult.action == Action.NONE) { return; }  // 如果无需动作, 则直接返回

    // 如果走到这里, 说明需要进行 SPILL 或 LOAD 动作. 首先确保动作不能太频繁, 如果小于阈值则直接返回, 不做动作
    // triggerInterval 的配置项为 state.backend.spillable.trigger-interval, 默认值为 1 分钟
    if (monitorResult.getTimestamp() - lastTriggerTime < triggerInterval) { return; }

    if (checkResult.action == Action.SPILL) {
        doSpill(checkResult);   // 调用 doSpill 方法将状态从内存移动到磁盘
    } else {
        doLoad(checkResult);    // 调用 doLoad 方法将状态从磁盘载入回内存
    }

    // spill 和 load 比较耗时, 所有事项做完后再更新 lastTriggerTime 时间
    lastTriggerTime = System.currentTimeMillis();
}

上述代码经过逐行注释后，逻辑非常清晰易懂。我们下面着重以三个核心方法 decideAction、doSpill、doLoad及其附属方法为例，介绍它们的工作流程。

decideAction 方法

这个方法根据 HeapStatusMonitor 返回的资源观测结果，做出实际的决策。

但是从下面的源码注释可以看到，目前决策的依据非常单一且不可靠，例如 Spill 只判断 GC 时间（这里计算的是近几次的平均时间，经常无法应对突发情况），Load 只判断内存用量等，而且不可动态调整，局限性非常大，因此这里也是迫切需要改进的一个点。

ActionResult decideAction(HeapStatusMonitor.MonitorResult monitorResult) {
    long gcTime = monitorResult.getGarbageCollectionTime();     // 获取近期 GC 平均时间
    long usedMemory = monitorResult.getTotalUsedMemory();       // 获取堆内存总用量

    // 1. 检查是否需要触发 SPILL
    if (gcTime > gcTimeThreshold) { // TODO: 目前只有 GC 时间是否超过阈值, 非常简陋
        return ActionResult.ofSpill(spillSizeRatio);    // TODO: 这个 ratio 需要改为动态的, 目前为配置项 state.backend.spillable.spill-size.ratio 设置, 默认值为 0.2, 表示每次有 20% 的状态被移动到磁盘上
    }

    // 2. 检查是否需要触发 LOAD
    if (usedMemory < loadStartSize) {   // TODO: 目前只有一个指标, loadStartSize = maxMemory * loadStartRatio, 而 loadStartRatio 由 state.backend.spillable.load-start.ratio 配置项决定, 默认是 0.1
        float loadRatio = (float) (loadEndSize - usedMemory) / usedMemory;  // loadEndSize = maxMemory * loadEndRatio, 而 loadEndRatio 由 state.backend.spillable.load-end.ratio 配置项决定, 默认为 0.3
        return ActionResult.ofLoad(loadRatio);  // 决定载入的内存比例
    }

    // 都不需要, 那么无动作
    return ActionResult.ofNone();
}

StateMapMeta

这个方法主要描述了在状态表中的一个 KeyGroup 所保存的所有状态的元数据信息，称为一个 StateMap。在 Flink 的状态系统中，分配的最小单元是 KeyGroup，表示哈希值一致的一组 Key。若干个连续的 KeyGroup 组成一个 KeyGroupRange，分配给相关算子的某个并行实例中。

public static class StateMapMeta {
    private final SpillableStateTable stateTable;    // KeyGroup 所属的状态表引用
    private final int keyGroupIndex;    // 该状态表中, 此 KeyGroup 的偏移量 (索引)
    private final boolean isOnHeap;     // 目前是否在堆内存里
    private final int size;             // 该 KeyGroup 的状态数
    private final long numRequests;     // 该 KeyGroup 的总请求数
    private long estimatedMemorySize;   // 估计的状态总大小, 如果是 -1 表示未初始化
}

可以看到，StateMapMeta 对象包括了这个 KeyGroup 所属对象的多种典型属性，Flink Spillable Backend 就是根据这些属性来计算权重、决定 Spill 还是 Load 等动作的影响范围。

getStateMapMetas 方法

这里我们简单介绍一下这个名为 getStateMapMetas 的辅助方法。它的作用是给定一个过滤器函数，对当前状态表里的所有 StateMap（即 KeyGroup 所属的状态集）进行筛选，然后返回符合条件的列表。

private List<SpillableStateTable.StateMapMeta> getStateMapMetas(
    Function<SpillableStateTable.StateMapMeta, Boolean> stateMapFilter) {   // 传入筛选策略函数, 按条件筛选状态表中符合条件的 KeyGroup 列表
    List<SpillableStateTable.StateMapMeta> stateMapMetas = new ArrayList<>();   // 创建一个列表, 以作为本函数返回值
    for (Tuple2<String, SpillableStateTable> tuple : stateTableContainer) {     // stateTableContainer 是即当前实例中, 所有已注册的状态名和状态表的 Tuple2 映射
        int len = stateMapMetas.size();
        SpillableStateTable spillableStateTable = tuple.f1;     // tuple.f1 是某个状态的状态表, 而 tuple.f0 是状态名, 这里用不到
        Iterator<SpillableStateTable.StateMapMeta> iterator = spillableStateTable.stateMapIterator();   // 准备按状态表中 KeyGroup 的顺序, 依次遍历检查该状态表里的所有的状态
        while (iterator.hasNext()) {    // 对该状态表中的每个 KeyGroup 下的状态映射进行遍历
            SpillableStateTable.StateMapMeta meta = iterator.next();
            if (stateMapFilter.apply(meta)) { stateMapMetas.add(meta); }    // 如果这个状态表的 KeyGroup 的 Meta 信息符合传入函数的筛选条件, 就加入返回列表
        }

        if (len < stateMapMetas.size()) {   // 如果发现上述循环中, 新增了符合条件的 KeyGroup, 需要估计表中每个 KeyGroup 的状态大小, 并写入元数据 (Meta) 中
            long estimatedSize = spillableStateTable.getStateEstimatedSize(true);   // 更新对象的平均大小, 注意是平均大小

            // 逐个更新每个新增的 KeyGroup 元数据中堆内存的估计大小, 计算公式是: 状态总数 * 估计的状态平均大小
            for (int i = len; i < stateMapMetas.size(); i++) {
                SpillableStateTable.StateMapMeta stateMapMeta = stateMapMetas.get(i);
                stateMapMeta.setEstimatedMemorySize(stateMapMeta.getSize() * estimatedSize);    // 状态总数 * 估计的状态平均大小
            }
        }
    }
    return stateMapMetas; // 返回含有最新状态估计大小的状态表元数据
}

特别需要注意的是，它的作用不仅仅在于筛选，而且还在筛选之后，对本 StateMap 里所有状态的大小进行估计，并保存在前面所述的 StateMapMeta 对象中。这样，后面的权重计算和排序，才有了数据支持。

doSpill 方法

下面我们来看一下，Spillable Backend 是如何将内存里的状态对象，Spill 到磁盘上的。这个 doSpill 方法由前述的 decideAction 方法调用，执行具体的 Spill 操作（优先选择访问不频繁、尺寸较大的 KeyGroup 进行 spill）。

它包括了筛选和大小估计、权重排序、执行 Spill 等多个步骤，最终达到阈值而完成整个流程。

void doSpill(ActionResult actionResult) {
    List<SpillableStateTable.StateMapMeta> onHeapStateMapMetas =    // 筛选当前 KeyGroup 并得到统计的元数据 (过滤条件是 onHeap 并且 size 大于 0)
        getStateMapMetas((meta) -> meta.isOnHeap() && meta.getSize() > 0);
    if (onHeapStateMapMetas.isEmpty()) { return; }  // 如果没有筛选到, 说明不用 spill, 直接返回

    sortStateMapMeta(actionResult.action, onHeapStateMapMetas);     // 根据 KeyGroup 的状态大小和访问频次进行权重排序, 权重大的放在前面

    long totalSize = onHeapStateMapMetas.stream()   // 获取所有堆内 KeyGroup 状态大小的总和
        .map(SpillableStateTable.StateMapMeta::getEstimatedMemorySize).reduce(0L, (a, b) -> a + b);     // 可以用 Long::sum 代替
    long spillSize = (long) (totalSize * actionResult.spillOrLoadRatio);    // 计算需要 spill 的比例

    if (spillSize == 0) { return; }
    if (cancelCheckpoint) { checkpointManager.cancelAllCheckpoints(); }     // 配置项 state.backend.spillable.cancel.checkpoint 可以控制是否在 spill 时取消当前的所有进行中的快照（默认为 true）。取消快照可以加快 GC 和 Spill 过程

    for (SpillableStateTable.StateMapMeta meta : onHeapStateMapMetas) {     // 根据排序得到的 KeyGroup 列表, 从权重最大（访问频次最小、大小最大）的开始, 逐个进行 Spill 操作, 直到达到阈值
        meta.getStateTable().spillState(meta.getKeyGroupIndex());
        spillSize -= meta.getEstimatedMemorySize();
        if (spillSize <= 0) { break; }  // 如果 Spill 的大小已经达到了阈值, 就不再继续, 本次 Spill 操作结束
    }
}

doLoad 方法

这个 doLoad 方法同样由 decideAction 方法调用，是 doSpill 方法的“对偶函数”，即将状态从磁盘等外存，载入到堆内存中（优先选择访问频繁、尺寸较小的 KeyGroup 进行 load）。

整体的函数逻辑与 doSpill 相同，只是更新阈值的代码放在了操作之前，以避免多载入对象到内存中，造成较大压力。

void doLoad(ActionResult actionResult) {    // 将状态从磁盘载入回堆内存
    List<SpillableStateTable.StateMapMeta> onDiskStateMapMetas =    // 筛选出所有不在堆内存且状态不为 0 的 KeyGroup 状态列表
        getStateMapMetas((meta) -> !meta.isOnHeap() && meta.getSize() > 0);
    if (onDiskStateMapMetas.isEmpty()) { return; }

    sortStateMapMeta(actionResult.action, onDiskStateMapMetas);     // 对所有的 KeyGroup 状态列表进行权重排序, 最大的 (访问次数最多、状态最小的）放在前面, 优先进行 Load

    long totalSize = onDiskStateMapMetas.stream()   // 计算符合条件的状态总大小
        .map(SpillableStateTable.StateMapMeta::getEstimatedMemorySize)
        .reduce(0L, Long::sum);
    long loadSize = (long) (totalSize * actionResult.spillOrLoadRatio); // 计算出需要载入的最大比例

    if (loadSize == 0) { return; }

    for (SpillableStateTable.StateMapMeta meta : onDiskStateMapMetas) { // 开始载入到内存, 直到满足阈值
        loadSize -= meta.getEstimatedMemorySize();
        // Load 时先减去状态大小, 避免多载入了一些状态, 导致内存压力比预期的更大
        if (loadSize < 0) { break; }

        meta.getStateTable().loadState(meta.getKeyGroupIndex());    // 按照 KeyGroup 元数据里面记录的 KeyGroupIndex 来载入状态到当前状态表
    }
}

sortStateMapMeta 方法

这个方法也属于比较关键的一个方法。它会根据传入的 StateMapMeta 列表，将其归一化以后，按照既定的规则进行权重排序。

需要注意的是，权重计算的方法在 computeWeight 方法中，根据 Spill（优先选择访问频率低、尺寸大的状态进行 Spill）还是 Load（优先选择访问频率高、尺寸小的状态进行 Load）有完全相反的权重计算方法。

private void sortStateMapMeta(Action action, List<SpillableStateTable.StateMapMeta> stateMapMetas) {
    if (stateMapMetas.isEmpty()) { return; }

    // 使用 (X - Xmin)/(Xmax - Xmin) 公式来进行归一化, 以确保归一化后的值域位于 [0,1]
    long sizeMax = 0L, sizeMin = Long.MAX_VALUE, requestMax = 0L, requestMin = Long.MAX_VALUE;
    for (SpillableStateTable.StateMapMeta meta : stateMapMetas) {   // 统计最大、最小的的 KeyGroup 状态对象总大小
        long estimatedMemorySize = meta.getEstimatedMemorySize();
        sizeMax = Math.max(sizeMax, estimatedMemorySize);
        sizeMin = Math.min(sizeMin, estimatedMemorySize);
        long numRequests = meta.getNumRequests();   // 获取该 KeyGroup 状态的总请求数
        requestMax = Math.max(requestMax, numRequests);
        requestMin = Math.min(requestMin, numRequests);
    }
    final long sizeDenominator = sizeMax - sizeMin; // 根据上述公式进行归一化
    final long requestDenominator = requestMax - requestMin;
    final long sizeMinForCompare = sizeMin;
    final long requestMinForCompare = requestMin;
    final Map<SpillableStateTable.StateMapMeta, Double> computedWeights = new IdentityHashMap<>();  // 准备一个 Map 来表示各个 KeyGroup 的相对权重
    Comparator<SpillableStateTable.StateMapMeta> comparator = (o1, o2) -> {     // 创建一个 KeyGroup 权重的 Comparator 用来排序
        if (o1 == o2) { return 0; }
        if (o1 == null) { return -1; }
        if (o2 == null) { return 1; }
        double weight1 = computedWeights.computeIfAbsent(o1,    // 计算第一个 KeyGroup 状态的权重, 并放入 computedWeights Map 中准备排序
            k -> computeWeight(k, action, sizeMinForCompare, requestMinForCompare, sizeDenominator, requestDenominator));   // computeWeight 的公式是 (weightRetainedSize * normalizedSize + weightRequestRate * normalizedRequest) / weightSum
        double weight2 = computedWeights.computeIfAbsent(o2,    // 计算第二个 KeyGroup 状态的权重, 并放入 computedWeights Map 中准备排序
            k -> computeWeight(k, action, sizeMinForCompare, requestMinForCompare, sizeDenominator, requestDenominator));
        // 对比权重, 然后对大的一方优先返回 -1, 这样在排序时会排到最前面, 优先进行 SPILL 或者 LOAD 等动作
        return (weight1 > weight2) ? -1 : 1;
    };
    stateMapMetas.sort(comparator); // 进行排序, 令大的 KeyGroup 可以出现在最前面
}

这个方法利用了 Java 自带的排序机制，通过自定义 Comparator 的方式，让权重最大的对象排在前面，这样构造了一个优先队列，前面介绍的doSpill 和 doLoad 方法都只需要从首部开始处理即可。

computeWeight 方法

这个 computeWeight 方法决定了排序时的先后顺序。可以看到，对于 SPILL 和 LOAD，计算的公式可以说是相反的，但是最终目的一致：根据概率原理，留在内存里的相对都是访问频繁及占空间较小的对象，有利于保持性能。

private double computeWeight(
    SpillableStateTable.StateMapMeta meta,
    Action action,
    long sizeMin, long requestMin,
    long sizeDenominator, long requestDenominator) {
    double normalizedSize = sizeDenominator == 0L ? 0.0 : (meta.getEstimatedMemorySize() - sizeMin) / (double) sizeDenominator; // 对本 KeyGroup 的状态大小进行归一化
    double normalizedRequest =  // 对本 KeyGroup 里状态的请求次数进行归一化
        requestDenominator == 0L ? 0.0 : (meta.getNumRequests() - requestMin) / (double) requestDenominator;
    double weightRetainedSize, weightRequestRate, weightSum;
    switch (action) {
        case SPILL: // 如果是 SPILL 操作, 倾向于选择访问不频繁的、较大的 Bucket 来进行
            weightRetainedSize = WEIGHT_SPILL_RETAINED_SIZE;    // 固定为 0.7, 正权重
            weightRequestRate = WEIGHT_SPILL_REQUEST_RATE;      // 固定为 -0.3, 负权重
            weightSum = WEIGHT_SPILL_SUM;   // 固定为前两者之和, 即 0.4
            break;
        case LOAD:  // 如果是 LOAD 操作, 倾向于选择请求频繁的、较小的 Bucket 来进行
            weightRetainedSize = WEIGHT_LOAD_RETAINED_SIZE; // 固定为 -0.3, 正权重
            weightRequestRate = WEIGHT_LOAD_REQUEST_RATE;   // 固定为 0.7, 负权重
            weightSum = WEIGHT_LOAD_SUM;    // 固定为前两者之和, 即 0.4
            break;
        default:
            throw new RuntimeException("Unsupported action: " + action);
    }
    // 如果是 Spill, 公式目前为 (0.7*normalizedSize-0.3*normalizedRequest)/0.4; 如果是 Load, 公式目前是 (0.7*normalizedRequest-0.3*normalizedSize)/0.4
    return (weightRetainedSize * normalizedSize + weightRequestRate * normalizedRequest) / weightSum; 
}

目前来看，公式里采用的参数都是固定的，对于调优来说并不是很灵活。

总结和展望

从目前的代码来看，SpillAndLoadManagerImpl 类的实现只能说完成了雏形，但是还有很多硬伤，例如公式参数固定不可调、决策条件过于简单、资源检测手段较为单一等等。

但需要看到，正是有了这些基石，该功能才有进一步完善的可能。因此这里非常感谢 Flink 社区的 Pengfei Li 和 Yu Li 两位作者的辛勤劳动。

我们也期待在接下来的版本中，Flink 的 Spillable Backend 可以更加成熟和完善；同时，后续我们也会将线上的一些经验和结论，融入到资源检测和动作判断逻辑中，让这个模块更加成熟、稳定，真正在生产环境下可用、可靠，最大程度上替代现有的 HeapKeyedStateBackend 和 RocksDBKeyedStateBackend。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

流计算 Oceanus