在 Flink 中使用 RocksDB 作为状态后端时，创建快照的 API 调用是什么？答案

【问题标题】：What are the API calls made to create snapshot while using RocksDB as the state backend in Flink?在 Flink 中使用 RocksDB 作为状态后端时，创建快照的 API 调用是什么？
【发布时间】：2020-07-26 17:13:27
【问题描述】：

我想分析 Flink 对 RocksDB 进行的每个 API 调用所花费的时间。但是，我找不到这些功能。

我曾尝试在 IDE 中设置 Flink 的完整源代码，将我的流式示例集成到源代码中，启动调试器并单步执行许多调用，但都是徒劳的。

示例如下：

package org.apache.flink.streaming.examples.spendreport;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.source.TransactionSource;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;

/**
 * Normal code.
 */
public class FraudDetectionAvi {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Transaction> transactions = env
            .addSource(new TransactionSource())
            .name("transactions");

        env.enableCheckpointing(60000);
        env.setStateBackend(new RocksDBStateBackend("file:///home/avsrivas/dev/flink/checkpoints", true));

        DataStream<Alert> alerts = transactions
            .keyBy(Transaction::getAccountId)
            .process(new FraudDetectorAvi())
            .name("fraud-detector");

        alerts
            .addSink(new AlertSink())
            .name("send-alerts");

        env.execute("Fraud Detection");
    }
}

点击here查看完整源代码。

我尝试单步执行，但无法推断调用 RocksDB 以保存状态的函数。

【问题讨论】：

标签： java apache-flink flink-streaming rocksdb

【解决方案1】：

当 RocksDB 用作 Flink 应用程序的状态后端时，任何键分区状态的工作副本都存储在每个任务管理器的本地嵌入式 RocksDB 实例中。计时器也可能保留在那里，或者它们可能在堆上。 RocksDB 将其状态保存在本地磁盘上；非键状态总是在堆上。

拍摄快照时（即在检查点期间或拍摄保存点时），存储在 RocksDB 中的状态会（异步）复制到快照存储（应该是分布式文件系统）。

在您的应用程序中，例如，当您调用 flagState.update(true) 时，会以 here, in RocksDBValueState.java 结尾，它使用此代码写入 RocksDB：

backend.db.put(columnFamily, writeOptions,
    serializeCurrentKeyWithGroupAndNamespace(),
    serializeValue(value));

稍后在快照期间会发生什么，取决于您是使用增量检查点还是完整检查点，但您会在 https://github.com/kebab-mai-haddi/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/snapshot 中找到 RocksDB 特定的代码。

请注意，快照不存储在 RocksDB 中。通过镜像 SST 文件来获取增量快照，而完整快照涉及迭代状态后端中的所有状态并写出结果。

有关 Flink 如何使用 RocksDB 的更多信息，请搜索 Stefan Richter 的博客文章和 Flink Forward 演讲。

【讨论】：