建网站建设网站,响应式网站微博视频教程,中明建投建设集团 网站,济宁市环保局建设项目审批网站本文主要分享与交流 Flink 状态使用过程中的一些经验与心得#xff0c;当然标题取了“最佳实践”之名#xff0c;希望文章内容能给读者带去一些干货。本文内容首先是回顾 state 相关概念#xff0c;并认识和区别不同的 state backend#xff1b;之后将分别对 state 使用访问…本文主要分享与交流 Flink 状态使用过程中的一些经验与心得当然标题取了“最佳实践”之名希望文章内容能给读者带去一些干货。本文内容首先是回顾 state 相关概念并认识和区别不同的 state backend之后将分别对 state 使用访问以及 checkpoint 容错相关内容进行详细讲解分享一些经验和心得。
State 概念回顾
我们先回顾一下到底什么是 state流式计算的数据往往是转瞬即逝 当然真实业务场景不可能说所有的数据都是进来之后就走掉没有任何东西留下来那么留下来的东西其实就是称之为 state中文可以翻译成状态。
在下面这个图中我们的所有的原始数据进入用户代码之后再输出到下游如果中间涉及到 state 的读写这些状态会存储在本地的 state backend可以对标成嵌入式本地 kv 存储当中。 接下来我们会在四个维度来区分两种不同的 stateoperator state 以及 keyed state。
1. 是否存在当前处理的 keycurrent keyoperator state 是没有当前 key 的概念而 keyed state 的数值总是与一个 current key 对应。2. 存储对象是否 on heap: 目前 operator state backend 仅有一种 on-heap 的实现而 keyed state backend 有 on-heap 和 off-heapRocksDB的多种实现。3. 是否需要手动声明快照snapshot和恢复 (restore) 方法operator state 需要手动实现 snapshot 和 restore 方法而 keyed state 则由 backend 自行实现对用户透明。4. 数据大小一般而言我们认为 operator state 的数据规模是比较小的认为 keyed state 规模是相对比较大的。需要注意的是这是一个经验判断不是一个绝对的判断区分标准。
StateBackend 的分类
下面这张图对目前广泛使用的三类 state backend 做了区分其中绿色表示所创建的operator/keyed state backend 是 on-heap 的黄色则表示是 off-heap 的。 一般而言在生产中我们会在 FsStateBackend 和 RocksDBStateBackend 间选择
FsStateBackend性能更好日常存储是在堆内存中面临着 OOM 的风险不支持增量 checkpoint。RocksDBStateBackend无需担心 OOM 风险是大部分时候的选择。
RocksDB StateBackend 概览和相关配置讨论
RocksDB 是 Facebook 开源的 LSM 的键值存储数据库被广泛应用于大数据系统的单机组件中。Flink 的 keyed state 本质上来说就是一个键值对所以与 RocksDB 的数据模型是吻合的。下图分别是 “window state” 和 “value state” 在 RocksDB 中的存储格式所有存储的 keyvalue 均被序列化成 bytes 进行存储。 在 RocksDB 中每个 state 独享一个 Column Family而每个 Column family 使用各自独享的 write buffer 和 block cache上图中的 window state 和 value state实际上分属不同的 column family。
下面介绍一些对 RocksDB 性能比较有影响的参数并整理了一些相关的推荐配置至于其他配置项可以参阅社区相关文档。
状态建议state.backend.rocksdb.thread.num后台 flush 和 compaction 的线程数. 默认值 ‘1‘. 建议调大state.backend.rocksdb.writebuffer.count每个 column family 的 write buffer 数目默认值 ‘2‘. 如果有需要可以适当调大state.backend.rocksdb.writebuffer.size每个 write buffer 的 size默认值‘64MB‘. 对于写频繁的场景建议调大state.backend.rocksdb.block.cache-size每个 column family 的 block cache大小默认值‘8MB’如果存在重复读的场景建议调大
State best practice一些使用 state 的心得
Operator state 使用建议
■ 慎重使用长 list
下图展示的是目前 task 端 operator state 在执行完 checkpoint 返回给 job master 端的 StateMetaInfo 的代码片段。 由于 operator state 没有 key group 的概念所以为了实现改并发恢复的功能需要对 operator state 中的每一个序列化后的元素存储一个位置偏移 offset也就是构成了上图红框中的 offset 数组。
那么如果你的 operator state 中的 list 长度达到一定规模时这个 offset 数组就可能会有几十 MB 的规模关键这个数组是会返回给 job master当 operator 的并发数目很大时很容易触发 job master 的内存超用问题。我们遇到过用户把 operator state 当做黑名单存储结果这个黑名单规模很大导致一旦开始执行 checkpointjob master 就会因为收到 task 发来的“巨大”的 offset 数组而内存不断增长直到超用无法正常响应。
■ 正确使用 UnionListState
union list state 目前被广泛使用在 kafka connector 中不过可能用户日常开发中较少遇到他的语义是从检查点恢复之后每个并发 task 内拿到的是原先所有operator 上的 state如下图所示 kafka connector 使用该功能为的是从检查点恢复时可以拿到之前的全局信息如果用户需要使用该功能需要切记恢复的 task 只取其中的一部分进行处理和用于下一次 snapshot否则有可能随着作业不断的重启而导致 state 规模不断增长。
Keyed state 使用建议
■ 如何正确清空当前的 state
state.clear() 实际上只能清理当前 key 对应的 value 值如果想要清空整个 state需要借助于 applyToAllKeys 方法具体代码片段如下 如果你的需求中只是对 state 有过期需求借助于 state TTL 功能来清理会是一个性能更好的方案。
■ RocksDB 中考虑 value 值很大的极限场景
受限于 JNI bridge API 的限制单个 value 只支持 2^31 bytes 大小如果存在很极限的情况可以考虑使用 MapState 来替代 ListState 或者 ValueState因为RocksDB 的 map state 并不是将整个 map 作为 value 进行存储而是将 map 中的一个条目作为键值对进行存储。
■ 如何知道当前 RocksDB 的运行情况
比较直观的方式是打开 RocksDB 的 native metrics 在默认使用 Flink managed memory 方式的情况下state.backend.rocksdb.metrics.block-cache-usage state.backend.rocksdb.metrics.mem-table-flush-pendingstate.backend.rocksdb.metrics.num-running-compactions 以及 state.backend.rocksdb.metrics.num-running-flushes 是比较重要的相关 metrics。
下面这张图是 Flink-1.10 之后打开相关 metrics 的示例图 而下面这张是 Flink-1.10 之前或者关闭 state.backend.rocksdb.memory.managed 的效果 ■ 容器内运行的 RocksDB 的内存超用问题
在 Flink-1.10 之前由于一个 state 独占若干 write buffer 和一块 block cache所以我们会建议用户不要在一个 operator 内创建过多的 state否则需要考虑到相应的额外内存使用量否则容易造成在容器内运行时相关进程被容器环境所杀。对于用户来说需要考虑一个 slot 内有多少 RocksDB 实例在运行一个 RocksDB 中有多少 state整体的计算规则就很复杂很难真得落地实施。
Flink-1.10 之后由于引入了 RocksDB 的内存托管机制在绝大部分情况下 RocksDB 的这一部分 native 内存是可控的不过受限于 RocksDB 的相关 cache 实现限制这里暂不展开后续会有文章讨论在某些场景下无法做到完美控制这时候建议打开上文提到的 native metrics观察相关 block cache 内存使用是否存在超用情况可以将相关内存添加到 taskmanager.memory.task.off-heap.size 中使得 Flink 有更多的空间给 native 内存使用。
一些使用 checkpoint 的使用建议
■ Checkpoint 间隔不要太短
虽然理论上 Flink 支持很短的 checkpoint 间隔但是在实际生产中过短的间隔对于底层分布式文件系统而言会带来很大的压力。另一方面由于检查点的语义所以实际上 Flink 作业处理 record 与执行 checkpoint 存在互斥锁过于频繁的 checkpoint可能会影响整体的性能。当然这个建议的出发点是底层分布式文件系统的压力考虑。
■ 合理设置超时时间
默认的超时时间是 10min如果 state 规模大则需要合理配置。最坏情况是分布式地创建速度大于单点job master 端的删除速度导致整体存储集群可用空间压力较大。建议当检查点频繁因为超时而失败时增大超时时间。
原文链接 本文为云栖社区原创内容未经允许不得转载。
