There is no golden egg in the RocksDB tuning area but it can help a significantly.
As a general rule the RocksDB state backend requires mainly 2 things to perform well, memory and fast local storage. You can tune these depending on the bottleneck and make measurements. Memory is used as a cache and the available amount will directly impact read/write throughput. RocksDB however does not use JVM Heap memory but instead it requires Flink managed memory. There are different ways to configure it but the recommended way is as a percentage of total TM memory. Adding more ephemeral storage is an obvious k8s config. G On Mon, Aug 19, 2024 at 4:21 PM Bjarke Tornager <bjarketorna...@gmail.com> wrote: > Hi Gabor, > > Thanks for the suggestions. > > For the rocksDB config tweaks would you be able to make some concrete > suggestions? I have looked into the official Tuning Checkpoints and Large > State guide documentation, however tuning rocksDB does not seem > straightforward tbh. > > My entire flinkConfiguration looks like this: > > spec: > flinkConfiguration: > env.java.opts.all: --add-opens java.base/java.lang=ALL-UNNAMED > --add-opens java.base/java.util=ALL-UNNAMED > --add-opens java.base/java.time=ALL-UNNAMED --add-opens > java.base/java.nio=ALL-UNNAMED > execution.checkpointing.file-merging.enabled: "true" > execution.checkpointing.interval: "120000" > execution.checkpointing.min-pause: "60000" > execution.checkpointing.timeout: "900000" > execution.checkpointing.unaligned: "true" > fs.default-scheme: abfss://<container-name>@<storage-account-name>. > dfs.core.windows.net > high-availability.storageDir: > abfss://<container-name>@<storage-account-name>. > dfs.core.windows.net/test/ha > high-availability.type: kubernetes > job.autoscaler.enabled: "true" > job.autoscaler.memory.tuning.enabled: "true" > job.autoscaler.metrics.window: 10m > job.autoscaler.stabilization.interval: 5m > job.autoscaler.target.utilization: "0.6" > job.autoscaler.target.utilization.boundary: "0.2" > pipeline.max-parallelism: "4" > rest.profiling.enabled: "true" > state.backend: rocksdb > state.backend.incremental: "true" > state.backend.rocksdb.checkpoint.transfer.thread.num: "8" > state.backend.rocksdb.compaction.style: UNIVERSAL > state.backend.rocksdb.localdir: /mnt/flink > state.backend.rocksdb.thread.num: "8" > state.backend.rocksdb.writebuffer.size: 128MB > state.checkpoints.dir: abfss://<container-name>@<storage-account-name>. > dfs.core.windows.net/test/checkpointing > state.savepoints.dir: abfss://<container-name>@<storage-account-name>. > dfs.core.windows.net/test/savepoints > taskmanager.network.memory.buffer-debloat.enabled: "true" > taskmanager.numberOfTaskSlots: "1" > > Best regards, > Bjarke > > > On Mon, Aug 19, 2024 at 1:18 PM Gabor Somogyi <gabor.g.somo...@gmail.com> > wrote: > >> Please see my comments inline. >> >> G >> >> >> On Thu, Aug 8, 2024 at 3:41 PM Bjarke Tornager <bjarketorna...@gmail.com> >> wrote: >> >>> Hi Gabor, >>> >>> What kind of development work - is it the Disaggregated State Storage >>> and Management FLIP ( >>> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=293046855) >>> that you are referring to? >>> >> Yes. >> >>> >>> Thanks for the suggestion. I am not familiar with the state processor >>> api but I will have a look. >>> >>> My state is not even that large(10-50 Gb). It seems like people are >>> using RocksDB with Tb of state but I am a bit puzzled as to how they handle >>> restarts from savepoint when it takes so long. Do you know? >>> >> Yeah, we know users with Tb of states and not easy to handle such jobs >> because of the long restore times. >> Practically the cluster must be strong enough to catch up the accumulated >> events after the long restore time + some config tweaks which I've listed >> in the next bullet point. >> >>> >>> Are there some flink configurations that I might be able to try out to >>> alleviate the slow restart that might be causing the exception? >>> >> What helps is "state.backend.local-recovery=true", some rocksDB tuning >> and reducing state size if possible (better eviction, etc...). >> >>> >>> Best regards, >>> Bjarke >>> >>> On Thu, Aug 8, 2024 at 11:16 AM Gabor Somogyi <gabor.g.somo...@gmail.com> >>> wrote: >>> >>>> Hi Bjarke, >>>> >>>> It's normal to see longer recovery time as the state size grows. There >>>> are developments in progress to mitigate this problem. >>>> >>>> > Any ideas as to what could be causing this? >>>> I think there is no easy way to tell it, I can give just some pointers. >>>> >>>> First I would take a look at the state files with the State Processor >>>> API[1], I would be really surprised though since ADLS is consistent, >>>> only the rename with overwrite operation is not atomic what we don't do >>>> in Flink. >>>> >>>> When the checkpoint files are not corrupt then I would add some debug >>>> printouts to the exception catch area and try to find out the >>>> exact root cause (since several Gb state can fit into memory maybe >>>> local repro can be possible with debugger). >>>> >>>> [1] >>>> https://nightlies.apache.org/flink/flink-docs-master/docs/libs/state_processor_api/ >>>> >>>> BR, >>>> G >>>> >>>> >>>> On Wed, Aug 7, 2024 at 9:46 PM Bjarke Tornager < >>>> bjarketorna...@gmail.com> wrote: >>>> >>>>> Hi, >>>>> >>>>> I am doing stateful streaming (keyed coprocessing) using rocksDB. When >>>>> my job restarts to adjust parallelism, and restores from savepoint, I >>>>> quite >>>>> often get the following error: >>>>> >>>>> 2024-08-07 19:43:09 >>>>> java.lang.Exception: Exception while creating >>>>> StreamOperatorStateContext. >>>>> at org.apache.flink.streaming.api.operators. >>>>> StreamTaskStateInitializerImpl.streamOperatorStateContext( >>>>> StreamTaskStateInitializerImpl.java:330) >>>>> at org.apache.flink.streaming.api.operators.AbstractStreamOperator >>>>> .initializeState(AbstractStreamOperator.java:275) >>>>> at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain >>>>> .initializeStateAndOpenOperators(RegularOperatorChain.java:106) >>>>> at org.apache.flink.streaming.runtime.tasks.StreamTask >>>>> .restoreStateAndGates(StreamTask.java:858) >>>>> at org.apache.flink.streaming.runtime.tasks.StreamTask >>>>> .lambda$restoreInternal$5(StreamTask.java:812) >>>>> at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1 >>>>> .call(StreamTaskActionExecutor.java:55) >>>>> at org.apache.flink.streaming.runtime.tasks.StreamTask >>>>> .restoreInternal(StreamTask.java:812) >>>>> at org.apache.flink.streaming.runtime.tasks.StreamTask.restore( >>>>> StreamTask.java:771) >>>>> at org.apache.flink.runtime.taskmanager.Task >>>>> .runWithSystemExitMonitoring(Task.java:970) >>>>> at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task >>>>> .java:939) >>>>> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) >>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) >>>>> at java.base/java.lang.Thread.run(Unknown Source) >>>>> Caused by: org.apache.flink.util.FlinkException: Could not restore >>>>> keyed state backend for >>>>> KeyedCoProcessOperator_30465cf1c06f2daf053ed603d38be4b6_(1/2) from any >>>>> of the 1 provided restore options. >>>>> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure >>>>> .createAndRestore(BackendRestorerProcedure.java:165) >>>>> at org.apache.flink.streaming.api.operators. >>>>> StreamTaskStateInitializerImpl.keyedStatedBackend( >>>>> StreamTaskStateInitializerImpl.java:457) >>>>> at org.apache.flink.streaming.api.operators. >>>>> StreamTaskStateInitializerImpl.streamOperatorStateContext( >>>>> StreamTaskStateInitializerImpl.java:203) >>>>> ... 12 more >>>>> Caused by: org.apache.flink.runtime.state.BackendBuildingException: >>>>> Caught unexpected exception. >>>>> at org.apache.flink.contrib.streaming.state. >>>>> RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder >>>>> .java:459) >>>>> at org.apache.flink.contrib.streaming.state. >>>>> EmbeddedRocksDBStateBackend.createKeyedStateBackend( >>>>> EmbeddedRocksDBStateBackend.java:533) >>>>> at org.apache.flink.contrib.streaming.state. >>>>> EmbeddedRocksDBStateBackend.createKeyedStateBackend( >>>>> EmbeddedRocksDBStateBackend.java:96) >>>>> at org.apache.flink.streaming.api.operators. >>>>> StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$3( >>>>> StreamTaskStateInitializerImpl.java:446) >>>>> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure >>>>> .attemptCreateAndRestore(BackendRestorerProcedure.java:173) >>>>> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure >>>>> .createAndRestore(BackendRestorerProcedure.java:137) >>>>> ... 14 more >>>>> Caused by: java.io.EOFException >>>>> at java.base/java.io.DataInputStream.readFully(Unknown Source) >>>>> at java.base/java.io.DataInputStream.readFully(Unknown Source) >>>>> at org.apache.flink.api.common.typeutils.base.array. >>>>> BytePrimitiveArraySerializer.deserialize(BytePrimitiveArraySerializer >>>>> .java:82) >>>>> at org.apache.flink.runtime.state.restore. >>>>> FullSnapshotRestoreOperation$KeyGroupEntriesIterator.next( >>>>> FullSnapshotRestoreOperation.java:298) >>>>> at org.apache.flink.runtime.state.restore. >>>>> FullSnapshotRestoreOperation$KeyGroupEntriesIterator.next( >>>>> FullSnapshotRestoreOperation.java:273) >>>>> at org.apache.flink.contrib.streaming.state.restore. >>>>> RocksDBFullRestoreOperation.restoreKVStateData( >>>>> RocksDBFullRestoreOperation.java:148) >>>>> at org.apache.flink.contrib.streaming.state.restore. >>>>> RocksDBFullRestoreOperation.applyRestoreResult( >>>>> RocksDBFullRestoreOperation.java:128) >>>>> at org.apache.flink.contrib.streaming.state.restore. >>>>> RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java: >>>>> 102) >>>>> at org.apache.flink.contrib.streaming.state. >>>>> RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder >>>>> .java:376) >>>>> ... 19 more >>>>> >>>>> However, most often the job is able to start up after several retries. >>>>> It seems to happen more often with higher parallelism (and higher cpu) and >>>>> as such with more pods. >>>>> Additionally, as the state grows into the Gb size it takes longer for >>>>> the job to start up and as such my job has longer downtime due to the many >>>>> retries before it successfully restores state. >>>>> >>>>> Any ideas as to what could be causing this? >>>>> >>>>> I am using Apache Flink Operator 1.8, Flink 1.20 using the official >>>>> Apache Flink image, Java 17, and Adls Gen2 for durable storage. >>>>> >>>>> Best regards, >>>>> Bjarke >>>>> >>>>
