关于后端:Flinkstate-的优化与-remotestate-的探索

摘要：本文整顿自 bilibili 资深开发工程师张杨，在 Flink Forward Asia 2022 核心技术专场的分享。本篇内容次要分为四个局部：

1. 相干背景
2. state 压缩优化
3. Remote state 摸索
4. 将来布局

点击查看原文视频 & 演讲 PPT

一、相干背景

1.1 业务详情

• 从业务规模来讲，B 站目前大概是 4000+ 的 Flink 工作，其中 95% 是 SQL 类型。
• 从部署模式来讲，B 站有 80% 的部署是 on yarn application 部署，咱们的 yarn 集群和离线的 yarn 是离开的，是实时专用的 yarn 集群。剩下的 20% 作业，为了响应公司降本增效的号召，目前是在线集群混部。这个计划的采纳次要是从老本思考，目前在应用 yarn on docker。

从 state 应用状况来讲，平台默认开启的全副是 rocksdb statebackend，大略有 50% 的工作都是带状态的，其中有上百个工作的状态超过了 500GB，这种工作咱们称之为超大状态的 state 工作。

1.2 statebackend 痛点

次要痛点有 3 个：

• cpu 抖动大。rocksdb 的原理决定了它肯定会有一个 compaction 的操作。目前，在小状态下这种操作对工作自身没什么影响，然而在大状态下，尤其是超过 500GB 左右，compaction 会造成 cpu 抖动大，峰值资源要求高。尤其是在 5、10、15 分钟这种 checkpoint 的整数被距离的状况下，整个 cpu 的起伏会比拟高。对于一个工作来说，它的峰值资源要求会比拟高。如果咱们的资源满足不了峰值资源的需要，工作就会触发一阵阵的沉积和消积的状况。
• 复原工夫长。因为 rocksdb 是一个嵌入式的 DB 存储，每次重启或者工作的 rescale 状态复原工夫长。因为它要把很大的文件下载下来，而后再在本地进行一些从新的 reload，这就须要一些大量的扫描工作，所以会导致整个工作的状态复原工夫比拟长。察看下来，失常来说大工作根本须要 5 分钟左右能力复原安稳的状态。
• rocksdb 十分依赖本地磁盘。rocksdb 是嵌入式的 DB，所以会十分依赖本地 io 的能力。因为往年公司有降本增效的策略，咱们也会在线举荐一些混部，因为 Flink 的带状态工作有对本地 io 依赖的特点，对于混部来讲是没法用的，这就导致咱们应用混部的资源是十分无限的。

次要通过两个方面解决这些痛点:

• 第一个是 state 压缩。出于对如何升高 compaction 开销的思考，对工作进行了一些 cpu 火焰图的剖析。通过剖析发现，整个 cpu 的耗费次要在 2 个方面，一个是 compaction，一个是压缩数据的换进换出，这就会带来大量的压缩和解压缩的工作。所以咱们在压缩层做了一些工作。
• 第二个是基于云原生的部署。因为本地没有 io 能力，咱们参考了一些云原生的计划，把整个本地的 rocksdb backend 进行了 Remote state 的降级。这相当于下面的一个存算拆散，而后近程进行了服务化的操作。这样就能够实现即连即用，不须要进行 reload 的操作，整个 state 是基于近程化进行的。

二、state 压缩优化

2.1 业务场景

从业务场景角度来看，大 state 剖析下来，它次要集中于模型训练和样本拼接。咱们的商业化和 AI 举荐部门就是这种典型的利用场景。它的特点就是 key 数据量宏大，散布稠密，很难命中 cache。且数据 ttl 短，根本都是小时级。

如下右图所示，整个业务的模型蕴含展示和点击。通过这两个动作咱们进行了校验，而后实现计算，从而生成样本，再灌入模型进行训练。这个场景下它的特点是缓存生效就很快生效，它的 key 数量比拟大且比拟稠密。

第二个点是数据的提交比拟短，那么整个缓存的生效会比拟快，这样就会导致 rocksdb 在应用的时候会重复从磁盘里一直地 reload 数据，而后进行压缩和解压缩，或者两头 compaction 数据一直地过期，导致 compaction 过于频繁。

2.2 优化思路

以上两个问题会导致整个 cpu 的耗费十分高，所以咱们能察看到，在后面提及的整个工作的峰值 cpu 耗费会十分高。咱们的优化思路是通过包含对社区的一些计划和业界其余公司分享的一些文章进行的调研，调研和优化计划次要围绕三个方面进行：

• 第一个是开启 partitoned index filter，缩小缓存竞争。从火焰图中能够看到，data block 会一直地从 block cache 中被换进换出，同时产生大量的压缩和解压缩工作，大量的 cpu 其实耗费在这里。
• 第二个是敞开 rocksdb 压缩，缩小缓存加载 /compaction 时候的 cpu 耗费。从火焰图中能够看到，整个 SSD 数据进行底层生成的时候，它会一直地进行整个文件的压缩和解压缩。所以咱们的思路就是如何去敞开一些 rocksdb 压缩，缩小整个缓存加载或者 cpu 的耗费。
• 第三个是反对接口层压缩，在数据 state 读写前后进行解压缩操作，缩小 state 大小。如果敞开 rocksdb 底层的压缩，咱们察看到整个磁盘的使用量十分高。这样的话，压力就是从压缩和解压缩的 cpu 耗费变成磁盘的 io 耗费，磁盘的 io 很容易就被耗费尽了。所以咱们就把整个底层的压缩往下层进行了一个牵引，反对接口层的压缩。

在数据层写入读写前后进行压缩和解压缩的操作，缩小底层的存储大小，咱们也与外部的一些 DB 存储团队进行了一些沟通，他们个别倡议在下层进行压缩，这样会比在底层做效率更高一些。因为底层整个是以 Block 维度进行压缩与解压缩的。

2.3 整体架构

partitoned index filter 开启，B 站外部目前次要用的是 Flink 1.11 版本。当初咱们大略调研到这个参数的时候，发现社区自身在 1.12 版本曾经晋升了这个性能，所以就间接把 1.12 版本相干的性能间接用过去了，同时 rocksdb 压缩敞开，官网文档均有参数，这个局部老本比拟低。

第二个局部是反对业务的压缩。为了缩小磁盘 io 的压力，在接口上做了一些压缩。接口压缩架构，如上图，重写了所有 rocksdb state 的 put/get 接口，应用的是最新的 Zstd 压缩算法。目前应用的是 LTZ 算法，比一些罕用的压缩收益更高，cpu 的耗费比拟小。

2.4 业务成果

如上业务成果比照图，这外面蕴含 test 1、2、3、4，次要看看 test 0 和 test 4。test 0 是社区默认的计划，也就是底层用是 snappy 压缩，下层不做什么。test 4 是齐全敞开了底层的压缩，无论是下层还是底层，整个压缩是全副敞开的，在业务层应用了最新的 STD 办法进行压缩。

整个察看下来，峰值和低峰期的数据都进行了查看，cpu 均值大略是降落到 15% 左右，峰值大略降落到 25%，置信对毛刺的降落会更显著一些。另外，在业务层做完压缩之后，这个文件大小根本放弃不变，或是只晋升了一点点。这样一来，最底层的存储空间或者 io 其实齐全没有减少任何压力。

目前来讲，线上所有新增的大的工作根本都曾经全副开启了，老的工作也在进行一些推动。

最初补充下这个场景的适应性。因为整个计划是匹配到咱们的举荐性能或是广告模型拼接场景来做的，所有的性能剖析也是在这样的场景下进行的，论断是次要双流 join 的大 state 场景，小 state 场景下根本没有收益，次要起因是小 state 的模式下，cache 自身的命中率十分高，换进换出频率较低。另外，Liststate 临时不反对业务层压缩，add 接口底层的非凡 merge 实现临时无奈做业务层压缩。

三、Remote state 摸索

3.1 业务场景和问题

第一，降本增效。这是往年大部分互联网公司可能都在做的事件，B 站也不列外。咱们在推一个离在线混部，次要是为了晋升资源的应用效率。把 Flink 大数据的场景和在线服务的场景进行资源的混部。

第二，大 state 作业重启下载 state 慢。有些用户反馈，对一些重启工夫比拟敏感的业务方，会感觉大状态下的重启对业务会有一些抖动，尤其在广告、AI 模型举荐等方面体验感不好。

第三，大 state 作业重启加载 state 慢。心愿在重启或是扩容的时候，加载更快一些。这里包含两个业务场景，一个是离在线混部，第二个是用户的重启体验。这两个方面即便是用 rocksdb 来做都比拟难。例如在线混部，因为在线的机器的磁盘根本是 100-200GB 的那种本地较小的磁板，同时 io 性能也不是很高，在这种状况下，将 Flink 大工作丢到在线混部下来，会对在线的机器负载压力比拟大。这就造成在线混部上的工作是绝对无限的。

其实用户很难辨别到底哪些工作是能够丢到在线混部，哪些不能。这须要用户通过 state 来判断。另外，state 的慢导致用户体验比拟差，次要是因为它重启的时候强依赖 load 过程，尤其是当遇到超过 500GB 的工作时候，这个体验感就会比拟差。

3.2 优化思路

首先是实现 Flink state 的存算拆散。如果进行混部，本地实际上没有磁盘，就很难给到在线的机器，在 Flink 上最间接的方法就是存算拆散。存算拆散不是生疏的概念，尤其是在云上。存储放在存储的机器上，依照存储的需要进行机型洽购，计算同理。

其次是 state 近程化 / 服务化，重启 /rescale 间接建设连贯进行读写。B 站的做法是把 Flink 的 state 进行存储的拆散，这样就能解决在线的混部问题。拆散之后，就能够进行近程化和服务化的部署，最终重启或是其余就能间接与近程服务进行连贯。

第三是近程 state 服务须要反对 checkpoint 残缺性能依赖的接口。这是失常开发流程中所依赖的，即近程服务。须要反对一个 checkpoint 的残缺性能，次要对标的就是 rocksdb 的性能。

3.3 整体架构

上图是 Flink state 整体的架构，也是 statebackend 对象拓扑，就是上图右侧。另外，Flink statebackend 的外围数据或是数据存储次要是由 keyed statebackend 承当，还有一部分是由 operator state 承当，这部分次要是存在内存里，并全量往近程刷，个别反对的数据不会特地大。

外围的大规模数据会存在 keyed statebackend 里。keyed statebackend 次要有 heap 和 rocksdb 两种实现，反对大规模额数据存储。

B 站次要应用的是 rocksdb 存储，次要围绕的外围是改良 keyed statebackend，因为 operator state 次要针对的是小规模数据，它不会占用本地 io，也不会在重启的时候产生大量的反复操作，速度相对来说要快很多。

上图是社区 keyed statebackend 的调配逻辑架构。如果想要把 keyed statebackend 做存算拆散，须要理解：

• Flink 的分片逻辑，它的底层有一个 keyed group 的概念。在所有状态在工作第一次启动的时候，底层分片就曾经确定好了。这个分片大小就是 Flink 的最大并发度的值。这个最大并发度实际上是在工作第一次不从 checkpoint 启动时候算进去的。这两头如果始终从 checkpoint 启动，那么最大并发度永远不会变。
• Flink checkpoint rescale 能力是无限的，实质是分片 key-group 在 subtask 之间的挪动，分片无奈决裂。一旦启动生成切换之后，这个分片的数量因为曾经固定好了，前面不管怎么 rescale，实质是分片的挪动，其实最大的变动是前面 rescale 不能超过整个分片的数量，这个数量对应的是分片最大的并发度。
• 多个 key-group 存储在雷同 rocksdb 的 cf 外面，以 key-group 的 ID 作为 key 的 prefix，依赖 rocksdb 的排序个性，rescale 过程通过 prefix 进行定位遍历提取。

上图是 Remote state 整体架构，次要的外围工作有三点：

• 替换 rocksdb 为 B 站外部的 taishan 存储。Taishan 存储底层是分布式构造，Flink 的一些概念是能够对应到 taishan 存储上的。
• key-group 概念对应 taishan 存储的 shard，cf 对应 table。key-group 的概念就能够对应上 taishan 存储，同时外面 cf 的概念也对应 taishan 存储中的一个 table 概念，也反对一个 table 外面能够有多个 key-group。
• Taishan 存储反对对应 shard 的 snapshot，以及从历史的 snapshot 进行疾速切换。比方重启的时候能够从老的 checkpoint 进行回复，实际上可能调用 taishan 存储这样一套接口进行疾速 snapshot 切换，并能够指定对应的 snapshot 并进行加载，加载的速度十分快，根本是秒级左右。

3.4 缓存架构

在上文介绍的根底上能够察看到，开发完之后实际上整个性能来说是齐全够用的，但也发现了一些比较严重的性能问题：

• 每次 state 操作都须要走网络 rpc，cpu 的耗费太高
• 网络 rpc 的提早高，工作吞吐低
• cache 是一个自然而然的抉择

3.5 缓存难点

写缓存绝对来讲是比较简单，内存攒批，后盾定期 checkpoint 刷进来就能够上百倍的缩小写的 rpc。

读缓存是比拟难的。读缓存是把场景进行拆分来看：

• key 比拟少的场景成果显著，命中率极高，成果好，甚至达到几百上千倍的成果晋升。
• 稠密的 key 场景，大量读 null，缓存命中率低。这个场景是当下比拟大的难点。
• 周期性业务导致缓存定期生效，读 null，命中率暴涨。周期性业务到周期末节点，就会触发一个缓存生效，一旦跨过这个节点，缓存里就会变成新的数据。如果刚巧遇上读 null，命中率就会很低，甚至导致工作抖动。这也是比拟大的难点。

基于以上的问题，优化计划是反对 key 全量缓存配置，无效解决稠密 / 缓存生效场景性能问题，使内存应用回升。然而这个优化毛病也比拟显著，即对内存的要求会高些。因为所有的 key 都要存在一个缓存里，而这个缓存就是内存。

除了以上介绍的两点难点，其实还有另外两个难点。其一是冷启动，因为缓存不会进行长久化，那么当工作重启等状态下，其实重启之后缓存内是空的，那么也就会造成重启的时候，读和写的申请量都很大，会给性能造成压力。

另外，全量 key 的缓存，启动的时候也没法立刻应用，须要等一个 ttl 周期，这样能力能够保障前面的全量 key 缓存是真正的全量 key。

3.6 off-heap 缓存模型图

上图是 off-heap 缓存模型，咱们应用的是 Facebook 的 OHC（off-heap cache）。能够察看到，基于 ttl 的被动淘汰，它带来的性能耗费绝对来讲会比拟大。然而如果齐全关掉基于 ttl 的淘汰机制，在某些非凡场景，基于 io 淘汰机制，也会耗费比拟大的空间。因为 io 淘汰的时候会进行判断哪个 key 能够真的被淘汰。如果齐全通过 io 淘汰，在有些场景上面可能断定了很多 key，这些 key 都无奈满足一个 io 的淘汰条件，那么这个淘汰流程就会很长，会导致接口卡的工夫会比拟久。所以在这里 B 站进行了调优，既保留了一个被动淘汰的思路，也依赖 io 这样一个淘汰机制。

目前来讲，咱们曾经实现针对局部工作的灰度，灰度工作的计算资源耗费升高了 30% 左右。近程 state 的服务耗费临时没有统计，粗略看整体的资源应该是持平或是只有稍微一点升高。离在线混部以及疾速 rescale 的意义是微小的。Flink 侧能够实现存算拆散的成果，同时，这也可能做疾速的重启而不须要进行状态的从新下载、从新加载或是 reload 的过程。

四、将来布局

点击查看原文视频 & 演讲 PPT

更多内容

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
0 元试用 实时计算 Flink 版（5000CU* 小时，3 个月内）
理解流动详情：https://click.aliyun.com/m/1000372333/