Yun, you mentioned that checkpoint also supports rescale. I thought the recommendation [1] is to use savepoint for rescale.
[1] https://www.ververica.com/blog/differences-between-savepoints-and-checkpoints-in-flink On Tue, May 26, 2020 at 6:46 AM Joey Pereira <j...@stripe.com> wrote: > Following up: I've put together the implementation, > https://github.com/apache/flink/pull/12345. It's passing tests but is > only partially complete, as it still needs some clean-up and configuration. > I still need to try running this against a production cluster to check the > performance, as well as getting some RocksDB benchmarks. > > On Mon, May 18, 2020 at 3:46 PM Joey Pereira <j...@stripe.com> wrote: > >> Thanks Yun for highlighting this, it's very helpful! I'll give it a go >> with that in mind. >> >> We have already begun using checkpoints for recovery. Having these >> improvements would still be immensely helpful to reduce downtime for >> savepoint recovery. >> >> On Mon, May 18, 2020 at 3:14 PM Yun Tang <myas...@live.com> wrote: >> >>> Hi Joey >>> >>> Previously, I also looked at the mechanism to create on-disk SSTables as >>> I planed to use RocksDB's benchmark to mock scenario in Flink. However, I >>> found the main challenge is how to ensure the keys are inserted in a >>> strictly increasing order. The key order in java could differ from the >>> bytes order in RocksDB. In your case, I think it could be much easier as >>> RocksFullSnapshotStrategy write data per columnfamily per key group which >>> should be in a strictly increasing order [1]. >>> >>> FLINK-17288 <https://issues.apache.org/jira/browse/FLINK-17288> could >>> mitigate the performance and your solution could help improve the >>> performance much better (and could integrate with state-processor-api >>> story). >>> >>> On the other hand, for out-of-box to use in production for your >>> scenario, how about using checkpoint to recover, as it also supports >>> rescale and normal recover. >>> >>> [1] >>> https://github.com/apache/flink/blob/f35679966eac9e3bb53a02bcdbd36dbd1341d405/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/snapshot/RocksFullSnapshotStrategy.java#L308 >>> >>> >>> Best >>> Yun Tang >>> ------------------------------ >>> *From:* Joey Pereira <j...@stripe.com> >>> *Sent:* Tuesday, May 19, 2020 2:27 >>> *To:* user@flink.apache.org <user@flink.apache.org> >>> *Cc:* Mike Mintz <mikemi...@stripe.com>; Shahid Chohan < >>> cho...@stripe.com>; Aaron Levin <aaronle...@stripe.com> >>> *Subject:* RocksDB savepoint recovery performance improvements >>> >>> Hey, >>> >>> While running a Flink application with a large-state, savepoint recovery >>> has been a painful part of operating the application because recovery time >>> can be several hours. During some profiling that chohan (cc'd) had done, a >>> red flag stood out — savepoint recovery consisted mostly of RocksDB Get and >>> Put operations. >>> >>> When Flink is bootstrapping state for RocksDB instances this is not what >>> I would have expected, as RocksDB supports direct ingestion of the on-disk >>> format (SSTables): >>> https://github.com/facebook/rocksdb/wiki/Creating-and-Ingesting-SST-files. >>> This >>> was also recently reported on Jira: >>> https://issues.apache.org/jira/browse/FLINK-17288. >>> >>> From what I understood of the current implementation: >>> >>> * The snapshot restoration pathways, RocksDBFullRestoreOperation and >>> RocksDBIncrementalRestoreOperation, >>> use RocksDBWriteBatchWrapper. >>> >>> * RocksDBWriteBatchWrapper is using RocksDB’s WriteBatch operator. This >>> will provide atomicity of batches as well as performance benefits for >>> batching, compared to individual Puts, but it will still involve RocksDB’s >>> insert paths which can involve expensive operations[0]. >>> >>> Instead, by creating SSTable files and instructing RocksDB to ingest the >>> files, writes can be batched even further and avoid expensive operations in >>> RocksDB. This is commonly utilized by other systems for restoration or >>> import processes, such as in CockroachDB[1], TiDB[2], and Percona[3]. There >>> are some restrictions on being able to generate SSTables, as well as >>> limitations for ingestion to be performant. Unfortunately, it’s all not >>> very well documented: >>> >>> 1. When generating an SSTable, keys need to be inserted in-order. >>> >>> 2. Ingested files should not have key-ranges that overlap with either >>> existing or other ingested files[4]. It is possible to ingest overlapping >>> SSTables, but this may incur significant overhead. >>> >>> To generate SSTables with non-overlapping key-ranges and to create them >>> with keys in-order, it would mean that the savepoints would need to be >>> ordered while processing them. I'm unsure if this is the case for how >>> Flink's savepoints are stored. >>> >>> I have not dug into RocksDBIncrementalRestoreOperation yet, or how it >>> is used (eg: for incremental checkpoint or something else). I did >>> notice it is iterating over a temporary RocksDB instance and inserting into >>> a "final” instance. These writes could be optimized in a similar >>> manner. Alternatively, it could be possible to use the temporary instance's >>> SSTables, ingest them, and prune data out with RocksDB's DeleteRange. >>> >>> To get started with prototyping, I was thinking of taking a simple >>> approach of making an interface for RocksDBWriteBatchWrapper and swapping >>> the implementation for one that does SSTable generation and ingestion. I >>> reckon that will be an easy way to sanity check whether it works at all. >>> >>> I was also planning on writing some benchmarks in RocksDB to understand >>> the difference for ingestion scenarios, as RocksDB itself is sparse on >>> details about SSTable ingestion[4] and does not have benchmarking for >>> ingestion. >>> >>> Does all of that seem sound? I'll report back when I get time to work >>> out that implementation and tests, likely during the coming weekend. >>> >>> >>> Joey >>> >>> --- >>> >>> [0]: I don’t have any specific sources on this. At a high-level, some of >>> the operations happening during writes include writing to the memtable >>> before flushing to an SSTable and doing merging and/or compaction. In >>> general, these will add write-amplification and overall overhead to bulk >>> insertion. These can largely be avoided by giving RocksDB SSTables, >>> especially if they have non-overlapping key-ranges. "Characterizing, >>> Modeling, and Benchmarking RocksDB Key-Value Workloads at Facebook" ( >>> https://www.usenix.org/system/files/fast20-cao_zhichao.pdf) is a >>> helpful source that highlights what happens during various workloads. >>> >>> >>> [1]: CockroachDB is a database that uses RocksDB as the on-disk storage. >>> Their implementation consolidates bulk ingestion to an AddSSTable >>> command. Primarily, they have some choice of options for SSTable generation >>> and ingestion that are of interest: >>> >>> * SSTable generation: >>> https://github.com/cockroachdb/cockroach/blob/c9aeb373511283db21b83c3c5a776ec2da2da1ed/c-deps/libroach/db.cc#L929-L966 >>> >>> * SSTable ingestion: >>> https://github.com/cockroachdb/cockroach/blob/c9aeb373511283db21b83c3c5a776ec2da2da1ed/c-deps/libroach/db.cc#L842-L917 >>> >>> >>> [2]: TiDB, and TiKV which is the KV layer of the database, uses RocksDB >>> as the on-disk storage. Their implementation of bulk ingestion is contained >>> within: >>> https://github.com/tikv/tikv/blob/master/components/sst_importer/src/sst_importer.rs >>> >>> Other useful references: >>> - https://github.com/tikv/tikv/issues/2404, discussing performance for >>> copy vs. move options. >>> >>> >>> [3]: Percona is a SQL database which supports a RocksDB backend. Their >>> implementation of ingestion can be found here: >>> https://github.com/percona/percona-server/blob/a259dc92e76e1569bc5ed34993cc7fc6af43832a/storage/rocksdb/ha_rocksdb.cc#L2815 >>> >>> >>> [4]: Again, there is not a lot of official resources on this. Notable >>> references I found on this include: >>> >>> * https://github.com/facebook/rocksdb/issues/2473, which describes at a >>> high-level how re-insertions work. >>> >>> * https://github.com/facebook/rocksdb/issues/3540, which describes the >>> performance costs for ingesting overlapping SSTables, and specific >>> benchmarks (post-fix) here: >>> https://github.com/facebook/rocksdb/pull/3564 >>> >>> * https://github.com/facebook/rocksdb/pull/3179, which describes the >>> mechanism for ingesting SSTable files: there need to be point-key overlap >>> checks for the LSM. >>> >>> * https://github.com/facebook/rocksdb/issues/5133, indicates >>> re-ingesting the same SSTable (due to restarts in import processes), >>> can cause issues for a particular set of options. >>> >>> * https://github.com/facebook/rocksdb/issues/5770#issuecomment-528488245, >>> indicates compaction occurs more (or, only) when overlapping SSTables >>> are ingested. The thinking here is non-overlapping SSTable ingestion means >>> very few operations (compaction, merging, etc) occur afterward, with >>> the right tuning for generation and ingestion. >>> >>> * https://github.com/facebook/rocksdb/issues/5010, which discusses some >>> unresolved issues for high CPU overhead on ingestion. >>> >>
