展开
关键词
腾讯云·社区
flink rocksdb
相关内容
流计算 Oceanus
流计算 Oceanus 是基于Flink构建的云上全托管的实时计算服务。您无须关注基础设施运维，通过云端一站式开发环境，轻松构建点击流分析、电商精准推荐、金融风控、物联网 IoT 等应用。
HOT
流计算 Oceanus 限时1元秒杀，1分钟快速构建轻量级实时 ETL 作业
如何在Apache Flink中管理RocksDB内存大小
，可以帮助我们有效地管理Apache Flink中RocksDB状态后端的内存大小。在之前的文章中，我们描述了Flink支持的状态后端选项。在这篇文章中，我们描述了RocksDB在Flink中的操作，然后我们介绍了一些有效资源消耗的重要配置。未来的文章将涵盖在Apache Flink中使用RocksDB进行额外调整，以便了解有关此主题的更多信息。Apache Flink中的RocksDB状态后端在深入了解配置参数之前，让我们首先重新讨论在flink中如何使用RocksDB来进行状态管理。我们刚刚引导您完成了一些用RocksDB作为Flink中的状态后端的的配置选项，这将帮助我们有效的管理内存大小。有关更多配置选项，我们建议您查看RocksDB调优指南或Apache Flink文档。Flink on RocksDB 参数调优指南
背景对于需要保存超大状态（远超于内存容量）的流计算场景来说，目前 RocksDB 是 Flink 平台上官方实现的唯一选择。因此我们对 Flink 上的 RocksDB 的参数调优方法进行了梳理，希望能够帮助大家解决相关的问题。Max Open Files这个参数决定了 RocksDB 可以打开的最大文件句柄数，在 Flink 的参数里是 state.backend.rocksdb.files.open.对于 Flink 的 RocksDB 衍生版，FRocksDB 而言，就是通过这种方式来实现内存容量的上限管理的。通常来说，Write Buffer 越大，写放大效应越小，因而写性能也会改善。总结RocksDB 提供了丰富的设置项，而 Flink 在此基础上封装为了自己的参数（大多可以通过配置项来设置，个别需要修改 StateBackend 初始化的代码），并将部分 RocksDB 的统计指标展示给用户Flink rocksdb如何做checkpoint
backend.db.put(columnFamily, writeOptions, rawKeyBytes, rawValueBytes);数据是存起来了，但是当进行 checkpoint 的时候， rocksdb这就要从 RocksDBKeyedStateBackend 说起了了，这个类很好的说明了 checkpoint 与 rocksdb 还有 hdfs 的关系当进行 checkpoint 的时候并且要对 keyedSnapshotDirectory snapshotDirectory = prepareLocalSnapshotDirectory(checkpointId); LOG.trace(Local RocksDBmeta data 主要就是一些元数据，像:rocksdb.column.family.name以及 state data的全目录。下面来看一下 state data： ?主要是 rocksdb 的一些配置信息以及当前的 db 目录， ?主要就是状态中(比如 mapState )存储的一些东西的具体。一文搞懂Flink rocksdb中的数据恢复
当我们设置 rocksdb state backend 时，并且从 checkpoint 重启时，首先进入 RocksDBKeyedStateBackendBuilder 的 getRocksDBRestoreOperationUUID.randomUUID().toString()); 首先呢会把 rawStateHandle 对应的 state 数据下载到 temporaryRestoreInstancePath 并且作为一个临时的 RocksDBgetOrRegisterStateColumnFamilyHandle( null, tmpRestoreDBInfo.stateMetaInfoSnapshots.get(i)) .columnFamilyHandle; 会把临时的 rockdb 实例的数据写入到 rocksdbRocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(iterator.key(), stopKeyGroupPrefixBytes)) { insert data to rocksdbApache Flink 管理大型状态之增量 Checkpoint 详解
如何使用当前，仅能够在 RocksDB StateBackend 上使用增量 checkpoint 机制，Flink 依赖 RocksDB 内部的备份机制来生成 checkpoint 文件。原理解析Flink 的增量 checkpoint 以 RocksDB 的 checkpoint 为基础。满了之后，RocksDB 会将所有数据以有序的写到磁盘。为此，Flink 调用 RocksDB 的 flush，强制将 memtable 的数据全部写到 sstable，并硬链到一个临时目录中。Flink 还能够保证所有的 checkpoint 都不会引用已经删除的文件，因为 RocksDB 中文件删除是由压缩完成的，压缩后会将原来的内容合并写成一个新的 sstable。Apache Flink 管理大型状态之增量 Checkpoint 详解
如何使用当前，仅能够在 RocksDB StateBackend 上使用增量 checkpoint 机制，Flink 依赖 RocksDB 内部的备份机制来生成 checkpoint 文件。以 RocksDB 的 checkpoint 为基础。满了之后，RocksDB 会将所有数据以有序的写到磁盘。为此，Flink 调用 RocksDB 的 flush，强制将 memtable 的数据全部写到 sstable，并硬链到一个临时目录中。Flink 还能够保证所有的 checkpoint 都不会引用已经删除的文件，因为 RocksDB 中文件删除是由压缩完成的，压缩后会将原来的内容合并写成一个新的 sstable。Managing Large State in Apache Flink®: An Intro to Incremental Checkpointing
uses RocksDB’s internal backup mechanism to consolidate checkpoint data over time.RocksDB is a key-value store based on ‘log-structured-merge’ (LSM) trees that collects all changes intracks which sstable files RocksDB has created and deleted since the previous checkpoint, and as theTo do this, Flink triggers a flush in RocksDB, forcing all memtables into sstables on disk, and hard-linkedThe columns in the figure above show the state of the local RocksDB instance for each checkpoint, theFlink CheckPoint奇巧 | 原理和在生产中的应用
现在只能通过RocksDB state back-end来获取增量式检查点的功能，Flink使用RocksDB内置的备份机制来合并检查点数据。在这个基础之上，Flink跟踪前一个checkpoint创建和删除的RocksDB sstable文件，因为sstable是不可变的，Flink可以因此计算出 状态有哪些改变。为了达到这个目标，Flink在RocksDB上触发了一个刷新操作，强制将memtable刷新到磁盘上。这个操作在Flink中是同步的，其他的操作是异步的，不会阻塞数据处理。checkpoint ‘CP2’，本地的RocksDB目录有两个sstable文件，这些文件是新生成的，于是Flink将它们传到了checkpoint 对应的存储目录。checkpoint‘CP2’,RocksDB有2个老的sstable文件,又创建了2个新的sstable文件。Flink将这两个新的sstable传到 持久化存储中，然后引用他们。Flink1.8.0重大更新-Flink中State的自动清除详解
在Flink 1.8.0中，该功能得到了扩展，包括对RocksDB和堆状态后端（FSStateBackend和MemoryStateBackend）的历史数据进行持续清理，从而实现旧条目的连续清理过程（由于上述这些限制，FLink应用程序仍需要在Flink 1.6.0中过期后主动删除状态。为了改善用户体验，Flink1.8.0引入了两种自主清理策略，分别针对Flink的两种状态后端类型。RocksDB后台压缩可以过滤掉过期状态如果你的Flink应用程序使用RocksDB作为状态后端存储，则可以启用另一个基于Flink特定压缩过滤器的清理策略。RocksDB定期运行异步压缩以合并状态更新并减少存储。Flink压缩过滤器使用TTL检查状态条目的到期时间戳，并丢弃所有过期值。激活此功能的第一步是通过设置以下Flink配置选项来配置RocksDB状态后端：state.backend.rocksdb.ttl.compaction.filter.enabled配置RocksDB状态后端后流计算 Oceanus
时间窗口函数,比较函数,类型转换函数,算术函数,聚合函数,逻辑和位运算函数,概览,命名规则,保留字,数据类型,术语表,流计算服务委托授权,CAM 访问管理,开发指南概述,CREATE FUNCTION,Flink配置项,查询语句,INSERT 语句,上下游数据一览,查看集群信息,作业信息,开发 SQL 作业,查看作业日志信息,作业操作,查看作业 Flink UI,配置作业监控告警,查看作业监控信息,依赖管理,聚合函数,逻辑和位运算函数,概览,标识符与保留字,命名规则,保留字,数据类型,术语和数据类型,术语表,流计算服务委托授权,操作指南,权限管理,CAM 访问管理,开发指南概述,CREATE FUNCTION,Flink配置项,DML 数据操作语句,查询语句,INSERT 语句,上下游开发指南,上下游数据一览,查看集群信息,作业管理,作业信息,开发 SQL 作业,查看作业日志信息,作业操作,查看作业 Flink UIFlink 清理过期 Checkpoint 目录的正确姿势
通过本文，你能 get 到以下知识：Flink Checkpoint 目录的清除策略生产环境应该选择哪种清除策略生产环境必须定期脚本清理 Checkpoint 和 Savepoint 目录RocksDB简单描述就是：这次故障是因为 Flink 基于 RocksDB 的增量 Checkpoint 导致。2.1 RocksDB 增量 Checkpoint 实现原理RocksDB 是一个基于 LSM 实现的 KV 数据库。图中所示的 merge 策略，完全依赖于 RocksDB 本身的合并策略，Flink 并不能主动控制。由于合并操作完全依赖 RocksDB，所以时间策略肯定不靠谱。Flink1.8新版发布:都有哪些改变
Flink 1.8引入了对RocksDB状态后端（FLINK-10471）和堆状态后端（FLINK-10473）的旧条数的连续清理。这意味着旧的条数将（根据TTL设置）不断被清理掉。在Flink附带的序列化程序中，我们现在支持PojoSerializer （FLINK-11485）和Java EnumSerializer （FLINK-11334）以及有限情况下的Kryo（FLINK可以通过升级到Flink 1.3和Flink 1.7之间的版本，然后再更新至Flink 1.8来解决此限制。4、RocksDB版本冲突并切换到FRocksDB（FLINK-10471）需要切换到名为FRocksDB的RocksDB的自定义构建，因为需要RocksDB中的某些更改来支持使用TTL进行连续状态清理FRocksDB的已使用版本基于RocksDB的升级版本5.17.2。对于Mac OS X，仅支持OS X版本> =10.13的RocksDB版本5.17.2。Flink 1.10 新特性研究
Flink 1.10 release 文档描述了一些比较重要的点，比如配置、操作、依赖、1.9 版本和 1.10 版本之间的区别，如果你准备将 Flink 升级到 1.10 版本，建议仔细看完下面的内容Mesos 会拒绝掉所有的过期请求•重构了 Flink 的调度程序，其目标是使调度策略在未来可以定制•支持 Java 11，当使用 Java 11 启动 Flink 时，会有些 WARNING 的日志提醒•RocksDB State Backend 内存可以控制，用户可以调整 RocksDB 的写读内存比率 state.backend.rocksdb.memory.write-buffer-ratio（升级，发布了自己的 FRocksDB（基于 RocksDB 5.17.2 版本），主要是因为高版本的 RocksDB 在某些情况下性能会下降•默认禁用 RocksDB 日志记录，需要启用的话需要利用可以在 flink-conf.yml 中修改 state.backend.rocksdb.write-batch-size 配置PyFlink•不再支持 Python2监控•InfluxdbReporterFlink 1.10 新特性研究
Flink 1.10 release 文档描述了一些比较重要的点，比如配置、操作、依赖、1.9 版本和 1.10 版本之间的区别，如果你准备将 Flink 升级到 1.10 版本，建议仔细看完下面的内容Mesos 会拒绝掉所有的过期请求•重构了 Flink 的调度程序，其目标是使调度策略在未来可以定制•支持 Java 11，当使用 Java 11 启动 Flink 时，会有些 WARNING 的日志提醒•RocksDB State Backend 内存可以控制，用户可以调整 RocksDB 的写读内存比率 state.backend.rocksdb.memory.write-buffer-ratio（升级，发布了自己的 FRocksDB（基于 RocksDB 5.17.2 版本），主要是因为高版本的 RocksDB 在某些情况下性能会下降•默认禁用 RocksDB 日志记录，需要启用的话需要利用可以在 flink-conf.yml 中修改 state.backend.rocksdb.write-batch-size 配置PyFlink•不再支持 Python2监控•InfluxdbReporterFlink1.8.0发布！新功能抢先看
Flink 1.8引入了对RocksDB状态后端（FLINK-10471）和堆状态后端（FLINK-10473）的旧条数的连续清理。这意味着旧的条数将（根据TTL设置）不断被清理掉。在Flink附带的序列化程序中，我们现在支持PojoSerializer（FLINK-11485）和Java EnumSerializer （FLINK-11334）以及有限情况下的Kryo（FLINK可以通过升级到Flink 1.3和Flink 1.7之间的版本，然后再更新至Flink 1.8来解决此限制。4、RocksDB版本冲突并切换到FRocksDB（FLINK-10471）需要切换到名为FRocksDB的RocksDB的自定义构建，因为需要RocksDB中的某些更改来支持使用TTL进行连续状态清理FRocksDB的已使用版本基于RocksDB的升级版本5.17.2。对于Mac OS X，仅支持OS X版本> =10.13的RocksDB版本5.17.2。Flink 使用大状态时的一点优化
常见的负载均衡策略Flink 源码中在选择 RocksDB 状态磁盘时，存在的问题。给出一些解决方案，并分析了每种方案利弊一、为什么要优化？使用多块硬盘来分担压力RocksDB 使用内存加磁盘的方式存储数据，当状态比较大时，磁盘占用空间会比较大。如果对 RocksDB 有频繁的读取请求，那么磁盘 IO 会成为 Flink 任务瓶颈。强烈建议在 flink-conf.yaml 中配置 state.backend.rocksdb.localdir 参数来指定 RocksDB 在磁盘中的存储目录。上述属于理想情况，当设置多个 RocksDB 本地磁盘目录时，Flink 会随机选择要使用的目录，所以就可能存在三个并行度共用同一目录的情况。Round Robin 策略轮循策略比较容易理解，多个 Worker 轮流接收数据即可，Flink 任务第一次申请 RocksDB 目录时使用目录1，第二次申请目录时使用目录2，依次申请即可。
相关视频
相关资讯
相关关键词
活动推荐
云+社区
扫码关注云+社区
领取腾讯云代金券