lgo opened a new pull request #14893: URL: https://github.com/apache/flink/pull/14893
## What is the purpose of the change Improves the performance of RocksDB state backend incremental rescales operations, where the state is being restored and clipped as part of the process. I piggybacked the previous ticket outlining this improvement that was since closed and partially done, let me know if a new ticket is preferred. This change was highlighted earlier by @sihuazhou in https://issues.apache.org/jira/browse/FLINK-8790 and https://github.com/apache/flink/pull/5582. Previously the actual change to use `deleteRange` was not done due to it being highlighted as experimental in this comment: https://github.com/apache/flink/pull/5582#issuecomment-391574919. `deleteRange` is in a far less experimental state now but I believe is still formally "experimental". It is heavily by many others like CockroachDB and TiKV and they have teased out several bugs in complex interactions over the years. ## Brief change log - Improves performance in certain re-scale situations when using the RocksDB state backend in incremental mode. ## Verifying this change This change is covered by existing tests, including: * `RocksDBIncrementalCheckpointUtilsTest` * `RocksIncrementalCheckpointRescalingTest` And, has an added test benchmark of `RocksIncrementalCheckpointRescalingBenchmarkTest`. I made further changes to the benchmark to generate numbers. These changes were omitted from this PR but are over in https://github.com/apache/flink/commit/3db9a8c4a1725c694bc734e99af3f2edf89f5312 on [lgo:joey-rescale-benchmark-changes](https://github.com/apache/flink/compare/master...lgo:joey-rescale-benchmark-changes). The change makes the benchmark side-by-side and run with a higher record count. Here were the results from several test runs with that commit on my local laptop. | Before | After | Improvement | | ------ | ----- | ----------- | | 2785.71ms | 2136.10ms | 23.3% | | 2962.27ms | 2221.91ms | 24.9% | | 2774.86ms | 2134.39ms | 23.0% | Here were the raw logs from the 3 benchmark executions. ``` [INFO] Running org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest 810 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> key group distribution <-------------- 816 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - [19939, 20206, 20101, 20164, 19978, 19987, 19969, 19943, 19881, 19832] 41033 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for all data: 2497184 41555 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=1 over begin=04 to end=05 42170 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=2 over begin=04 to end=05 42170 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=3 over begin=04 to end=05 42541 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=0 is: 1048929 44126 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=1 is: 753372 45262 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=2 is: 750889 45508 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> performance for incremental checkpoint re-scaling <-------------- 45508 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - useFastMode=false rescale from 2 to 3 with 200000 records took: 2785712528 nanoseconds 45521 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> key group distribution <-------------- 45521 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - [19939, 20206, 20101, 20164, 19978, 19987, 19969, 19943, 19881, 19832] 84545 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for all data: 2497200 85043 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=1 over begin=04 to end=05 85044 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=2 over begin=04 to end=05 85044 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=3 over begin=04 to end=05 85494 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=0 is: 1237730 87168 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=1 is: 753374 88280 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=2 is: 750891 88506 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> performance for incremental checkpoint re-scaling <-------------- 88506 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - useFastMode=true rescale from 2 to 3 with 200000 records took: 2136109710 nanoseconds [INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 87.726 s - in org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [INFO] Running org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest 810 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> key group distribution <-------------- 816 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - [19939, 20206, 20101, 20164, 19978, 19987, 19969, 19943, 19881, 19832] 43449 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for all data: 2497193 43969 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=1 over begin=04 to end=05 44729 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=2 over begin=04 to end=05 44729 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=3 over begin=04 to end=05 45096 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=0 is: 1053572 46742 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=1 is: 753372 47885 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=2 is: 750890 48120 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> performance for incremental checkpoint re-scaling <-------------- 48121 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - useFastMode=false rescale from 2 to 3 with 200000 records took: 2962274877 nanoseconds 48156 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> key group distribution <-------------- 48157 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - [19939, 20206, 20101, 20164, 19978, 19987, 19969, 19943, 19881, 19832] 89727 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for all data: 2497188 90224 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=1 over begin=04 to end=05 90225 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=2 over begin=04 to end=05 90225 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=3 over begin=04 to end=05 90670 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=0 is: 1237730 92433 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=1 is: 753373 93546 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=2 is: 750889 93774 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> performance for incremental checkpoint re-scaling <-------------- 93774 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - useFastMode=true rescale from 2 to 3 with 200000 records took: 2221911110 nanoseconds [INFO] Running org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest 794 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> key group distribution <-------------- 798 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - [19939, 20206, 20101, 20164, 19978, 19987, 19969, 19943, 19881, 19832] 40735 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for all data: 2497188 41251 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=1 over begin=04 to end=05 41838 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=2 over begin=04 to end=05 41838 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange slow mode for columnFamily=3 over begin=04 to end=05 42204 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=0 is: 1053376 43880 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=1 is: 753372 44990 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=2 is: 750890 45228 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> performance for incremental checkpoint re-scaling <-------------- 45228 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - useFastMode=false rescale from 2 to 3 with 200000 records took: 2774862604 nanoseconds 45238 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> key group distribution <-------------- 45238 [main] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - [19939, 20206, 20101, 20164, 19978, 19987, 19969, 19943, 19881, 19832] 83937 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for all data: 2497190 84446 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=1 over begin=04 to end=05 84447 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=2 over begin=04 to end=05 84447 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils [] - deleteRange fast mode for columnFamily=3 over begin=04 to end=05 84893 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=0 is: 1237730 86592 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=1 is: 753373 87692 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - state size for subtask=2 is: 750889 87928 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - --------------> performance for incremental checkpoint re-scaling <-------------- 87929 [Time-limited test] ERROR org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest [] - useFastMode=true rescale from 2 to 3 with 200000 records took: 2134395504 nanoseconds [INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 87.161 s - in org.apache.flink.contrib.streaming.state.RocksIncrementalCheckpointRescalingBenchmarkTest ``` ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): **no** - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no** - The serializers: **no** - The runtime per-record code paths (performance sensitive): **no** - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: **yes** - The S3 file system connector: **no** ## Documentation - Does this pull request introduce a new feature? (yes / no) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
