英文:
Flink RocksDB custom options factory config error disable block cache
问题
I am running Flink 1.15.2 and am trying to define a custom options factory in RocksDB to disable the block cache.
Following the example from this blog post: https://shopify.engineering/optimizing-apache-flink-applications-tips
However, my Flink application refuses to start after adding the OptionsFactory
to my env. The error seems to come from this line
https://github.com/facebook/rocksdb/blob/main/table/block_based/block_based_table_factory.cc#L599
Enable cache_index_and_filter_blocks, , but block cache is disabled
I have no idea what is gone wrong here, something must be overriding/ignoring my column options, but I don't know what.
Any help is much appreciated.
The config is here:
class NoBlockCacheRocksDbOptionsFactory extends ConfigurableRocksDBOptionsFactory {
override def createDBOptions(currentOptions: DBOptions, handlesToClose: util.Collection[AutoCloseable]): DBOptions = {
currentOptions.setMaxBackgroundJobs(20) // state.backend.rocksdb.thread.num
currentOptions
}
override def createColumnOptions(
currentOptions: ColumnFamilyOptions,
handlesToClose: util.Collection[AutoCloseable]): ColumnFamilyOptions = {
val blockBasedTableConfig = new BlockBasedTableConfig()
.setNoBlockCache(true)
.setBlockCache(null)
.setCacheIndexAndFilterBlocks(false)
.setCacheIndexAndFilterBlocksWithHighPriority(false)
.setPinL0FilterAndIndexBlocksInCache(false)
currentOptions.setTableFormatConfig(blockBasedTableConfig)
}
override def configure(configuration: ReadableConfig): RocksDBOptionsFactory = {
this
}
}
Error and stack trace:
Caused by: org.apache.flink.util.SerializedThrowable: Enable cache_index_and_filter_blocks, , but block cache is disabled
at org.rocksdb.RocksDB.open(Native Method) ~[flink-dist-1.15.2.jar:1.15.2]
at org.rocksdb.RocksDB.open(RocksDB.java:306) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.openDB(RocksDBOperationUtils.java:80) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.restore.RocksDBHandle.loadDb(RocksDBHandle.java:134) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.restore.RocksDBHandle.openDB(RocksDBHandle.java:113) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.restore.RocksDBNoneRestoreOperation.restore(RocksDBNoneRestoreOperation.java:62) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:315) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:483) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:97) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:329) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:346) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:164) ~[flink-dist-1.15.2.jar:1.15.2]
... 11 more
英文:
I am running Flink 1.15.2 and am trying to define a custom options factory in RocksDB to disable the block cache.
Following the example from this blog post: https://shopify.engineering/optimizing-apache-flink-applications-tips
However, my Flink application refuses to start after adding the OptionsFactory
to my env. The error seems to come from this line
https://github.com/facebook/rocksdb/blob/main/table/block_based/block_based_table_factory.cc#L599
Enable cache_index_and_filter_blocks, , but block cache is disabled
I have no idea what is gone wrong here, something must be overriding/ignoring my column options, but I don't know what.
Any help is much appreciated.
The config is here:
class NoBlockCacheRocksDbOptionsFactory extends ConfigurableRocksDBOptionsFactory {
override def createDBOptions(currentOptions: DBOptions, handlesToClose: util.Collection[AutoCloseable]): DBOptions = {
currentOptions.setMaxBackgroundJobs(20) // state.backend.rocksdb.thread.num
currentOptions
}
override def createColumnOptions(
currentOptions: ColumnFamilyOptions,
handlesToClose: util.Collection[AutoCloseable]): ColumnFamilyOptions = {
val blockBasedTableConfig = new BlockBasedTableConfig()
.setNoBlockCache(true)
.setBlockCache(null)
.setCacheIndexAndFilterBlocks(false)
.setCacheIndexAndFilterBlocksWithHighPriority(false)
.setPinL0FilterAndIndexBlocksInCache(false)
currentOptions.setTableFormatConfig(blockBasedTableConfig)
}
override def configure(configuration: ReadableConfig): RocksDBOptionsFactory = {
this
}
}
Error and stack trace:
Caused by: org.apache.flink.util.SerializedThrowable: Enable cache_index_and_filter_blocks, , but block cache is disabled
at org.rocksdb.RocksDB.open(Native Method) ~[flink-dist-1.15.2.jar:1.15.2]
at org.rocksdb.RocksDB.open(RocksDB.java:306) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.openDB(RocksDBOperationUtils.java:80) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.restore.RocksDBHandle.loadDb(RocksDBHandle.java:134) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.restore.RocksDBHandle.openDB(RocksDBHandle.java:113) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.restore.RocksDBNoneRestoreOperation.restore(RocksDBNoneRestoreOperation.java:62) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:315) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:483) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:97) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:329) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:346) ~[flink-dist-1.15.2.jar:1.15.2]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:164) ~[flink-dist-1.15.2.jar:1.15.2]
... 11 more
答案1
得分: 1
尝试将Flink配置中的state.backend.rocksdb.memory.managed从默认值true更改为false。此设置将覆盖作为RocksDB选项工厂的一部分设置的配置。禁用更改此值后,我能够成功运行作业。
英文:
try setting the flink config state.backend.rocksdb.memory.managed to false from its default value of true.
This setting override configs set as a part of the rocksdb options factory. I was able to run the job post disabling changing this value.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论