0
# Apache Flink RocksDB State Backend
1

2
Apache Flink RocksDB State Backend provides a persistent state backend implementation that uses RocksDB as the underlying storage engine. This enables streaming applications to maintain large amounts of keyed state that can survive job failures and restarts through checkpoints, with support for incremental checkpointing and customizable performance tuning.
3

4
## Package Information
5

6
- **Package Name**: org.apache.flink:flink-statebackend-rocksdb_2.12
7
- **Package Type**: maven
8
- **Language**: Java
9
- **Installation**: Add to your `pom.xml`:
10

11
```xml
12
<dependency>
13
    <groupId>org.apache.flink</groupId>
14
    <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
15
    <version>1.14.6</version>
16
</dependency>
17
```
18

19
## Core Imports
20

21
```java
22
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
23
import org.apache.flink.contrib.streaming.state.RocksDBOptions;
24
import org.apache.flink.contrib.streaming.state.PredefinedOptions;
25
import org.apache.flink.contrib.streaming.state.RocksDBOptionsFactory;
26
import org.apache.flink.contrib.streaming.state.DefaultConfigurableOptionsFactory;
27
import org.apache.flink.contrib.streaming.state.RocksDBNativeMetricOptions;
28
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackendFactory;
29
```
30

31
**Note**: The legacy `RocksDBStateBackend` class is deprecated as of Flink 1.13. Use `EmbeddedRocksDBStateBackend` for new applications.
32

33
## Basic Usage
34

35
```java
36
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
37
import org.apache.flink.contrib.streaming.state.PredefinedOptions;
38
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
39

40
// Create Flink execution environment
41
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
42

43
// Configure RocksDB state backend
44
EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend(true);
45
stateBackend.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED);
46
stateBackend.setDbStoragePath("/path/to/rocksdb/storage");
47

48
// Set state backend on environment
49
env.setStateBackend(stateBackend);
50

51
// Your Flink application logic...
52
```
53

54
## Architecture
55

56
The RocksDB State Backend is built around several key components:
57

58
- **EmbeddedRocksDBStateBackend**: Primary state backend class that manages local RocksDB instances for keyed state storage
59
- **Configuration System**: Comprehensive options through predefined configurations or custom factory implementations
60
- **Memory Management**: Integrates with Flink's managed memory system for optimal resource utilization
61
- **Checkpoint Integration**: Supports both full and incremental checkpointing for fault tolerance
62
- **Performance Tuning**: Extensive configuration options for different hardware profiles and workload characteristics
63

64
## Capabilities
65

66
### State Backend Configuration
67

68
Core functionality for creating and configuring the RocksDB state backend, including storage paths, incremental checkpointing, and basic setup options.
69

70
```java { .api }
71
class EmbeddedRocksDBStateBackend {
72
    EmbeddedRocksDBStateBackend();
73
    EmbeddedRocksDBStateBackend(boolean enableIncrementalCheckpointing);
74
    void setDbStoragePath(String path);
75
    void setDbStoragePaths(String... paths);
76
    void setPredefinedOptions(PredefinedOptions options);
77
    void setRocksDBOptions(RocksDBOptionsFactory optionsFactory);
78
}
79
```
80

81
[State Backend Configuration](./state-backend-configuration.md)
82

83
### RocksDB Options Factory
84

85
Factory pattern for customizing RocksDB database and column family options, enabling fine-grained performance tuning for specific use cases and hardware configurations.
86

87
```java { .api }
88
interface RocksDBOptionsFactory {
89
    DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose);
90
    ColumnFamilyOptions createColumnOptions(ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose);
91
    default RocksDBNativeMetricOptions createNativeMetricsOptions(RocksDBNativeMetricOptions nativeMetricOptions);
92
}
93

94
class DefaultConfigurableOptionsFactory implements RocksDBOptionsFactory, ConfigurableRocksDBOptionsFactory {
95
    DefaultConfigurableOptionsFactory setMaxBackgroundThreads(int totalThreadCount);
96
    DefaultConfigurableOptionsFactory setWriteBufferSize(String writeBufferSize);
97
    DefaultConfigurableOptionsFactory setBlockCacheSize(String blockCacheSize);
98
}
99
```
100

101
[Options Factory](./options-factory.md)
102

103
### Memory Configuration
104

105
Memory management configuration for RocksDB integration with Flink's managed memory system, including write buffer ratios and memory allocation strategies.
106

107
```java { .api }
108
class RocksDBMemoryConfiguration {
109
    void setUseManagedMemory(boolean useManagedMemory);
110
    void setFixedMemoryPerSlot(MemorySize fixedMemoryPerSlot);
111
    void setWriteBufferRatio(double writeBufferRatio);
112
    void setHighPriorityPoolRatio(double highPriorityPoolRatio);
113
    boolean isUsingManagedMemory();
114
}
115
```
116

117
[Memory Configuration](./memory-configuration.md)
118

119
### Predefined Options
120

121
Pre-configured RocksDB options optimized for different hardware profiles and use cases, providing easy setup for common deployment scenarios.
122

123
```java { .api }
124
enum PredefinedOptions {
125
    DEFAULT,
126
    SPINNING_DISK_OPTIMIZED,
127
    SPINNING_DISK_OPTIMIZED_HIGH_MEM,
128
    FLASH_SSD_OPTIMIZED;
129
    
130
    abstract DBOptions createDBOptions(Collection<AutoCloseable> handlesToClose);
131
    abstract ColumnFamilyOptions createColumnOptions(Collection<AutoCloseable> handlesToClose);
132
}
133
```
134

135
[Predefined Options](./predefined-options.md)
136

137
### Factory Configuration
138

139
Factory class for creating state backend instances from configuration, enabling integration with Flink's configuration system and deployment frameworks.
140

141
```java { .api }
142
class EmbeddedRocksDBStateBackendFactory {
143
    static EmbeddedRocksDBStateBackend createFromConfig(ReadableConfig config, ClassLoader classLoader);
144
}
145
```
146

147
### Native Metrics Configuration
148

149
Configuration for RocksDB native metrics monitoring, enabling detailed performance monitoring and observability of RocksDB internal operations.
150

151
```java { .api }
152
class RocksDBNativeMetricOptions {
153
    static RocksDBNativeMetricOptions fromConfig(ReadableConfig config);
154
    void enableNumImmutableMemTable();
155
    void enableCompactionPending();
156
    void enableBlockCacheUsage();
157
    void setColumnFamilyAsVariable(boolean columnFamilyAsVariable);
158
}
159
```
160

161
[Native Metrics Configuration](./native-metrics-configuration.md)
162

163
## Types
164

165
```java { .api }
166
enum EmbeddedRocksDBStateBackend.PriorityQueueStateType {
167
    HEAP,
168
    ROCKSDB
169
}
170

171
enum org.rocksdb.CompactionStyle {
172
    LEVEL,
173
    UNIVERSAL,
174
    FIFO,
175
    NONE
176
}
177

178
enum org.rocksdb.InfoLogLevel {
179
    DEBUG_LEVEL,
180
    INFO_LEVEL,
181
    WARN_LEVEL,
182
    ERROR_LEVEL,
183
    FATAL_LEVEL,
184
    HEADER_LEVEL
185
}
186
```