在Flink 1.13中配置RocksDB

我读过有关Flink 1.13版本中EmbeddedRocksDBStateBackend的内容,但有大小限制,因此我希望保留以前Flink版本1.11的当前配置,但重点是这种配置RocksDB的方式已被弃用(new RocksDBStateBackend("path", true);)。

我已使用EmbeddedRocksDBStateBackend (new EmbeddedRocksDBStateBackend(true))尝试使用新配置,但出现以下错误:

java.util.concurrent.ExecutionException: java.io.IOException: Size of the state is larger than the maximum permitted memory-backed state. Size=9126648 , maxSize=5242880 . Consider using a different state backend, like the File System State backend.

从Java以编程方式为Flink 1.13配置RocksDB状态后端的最佳方式是什么?


解决方案

在Flink1.13中,我们重新组织了状态后端,因为旧的方法导致了对事物如何工作的许多误解。因此,这两个问题是分离的:

  1. 您的工作状态存储在哪里(状态后端)。(对于RocksDB,应将其配置为使用最快的可用本地磁盘。)
  2. 存储检查点的位置(检查点存储)。在大多数情况下,这应该是分布式文件系统。
在旧的API中,通过将检查点路径传递给RocksDBStateBackend构造函数的方式,掩盖了在RocksDB的情况下涉及两个不同文件系统的事实。因此该配置位已移至其他位置(见下文)。

此表显示旧状态后端与新状态后端(与检查点存储结合使用)之间的关系:

传统状态后端 新状态后端+检查点存储
MemoryStateBackend HashMapStateBackend + JobManagerCheckpointStorage
FsStateBackend HashMapStateBackend + FileSystemCheckpointStorage
RocksDBStateBackend EmbeddedRocksDBStateBackend + FileSystemCheckpointStorage
在您的案例中,您希望将EmbeddedRocksDBStateBackendFileSystemCheckpointStorage一起使用。您当前遇到的问题是,您正在对RocksDB使用内存中检查点存储(JobManagerCheckpointStorage),这会严重限制可以设置检查点的状态数量。

您可以通过在flink-conf.yaml

中指定检查点目录来修复此问题
state.backend: rocksdb
state.checkpoints.dir: file:///checkpoint-dir/

# Optional, Flink will automatically default to FileSystemCheckpointStorage
# when a checkpoint directory is specified.
state.checkpoint-storage: filesystem

或在您的代码中

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");

// If you manually passed FsStateBackend into the RocksDBStateBackend constructor
// to specify advanced checkpointing configurations such as write buffer size,
// you can achieve the same results by using manually instantiating a FileSystemCheckpointStorage object.
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"));

有关完整详细信息,请参阅Migrating from Legacy Backends上的文档。

相关文章