使用 Cassandra 编写检查点答案

【问题标题】：Checkpointing with Cassandra write使用 Cassandra 编写检查点
【发布时间】：2018-04-12 11:20:10
【问题描述】：

当在写入 Cassandra 的 Flink 作业中使用检查点时，当此写入由于连接问题而失败时，作业会失败并在一定时间间隔后重新启动。

当记录失败时，该作业从哪里开始？它是选择下一条记录来处理还是重置偏移量并尝试重新处理失败的记录？

我的检查点配置如下所示，

try{
        env.setStateBackend(new RocksDBStateBackend(props.getFlinkCheckpointDataUri(), true));
        env.enableCheckpointing(10000, EXACTLY_ONCE); //10 seconds

        CheckpointConfig config = env.getCheckpointConfig();
        config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
    }
catch(Exception e){
        System.out.println("Failed to prepare stream execution environment");
    }

【问题讨论】：

标签： cassandra apache-flink flink-streaming

【解决方案1】：

如果启用了检查点，Flink 保证 Cassandra 接收器至少交付一次（如果您对 C* 实例的更新请求是幂等的，则完全一次交付；这意味着可以多次应用更新而不改变结果）[ref] .换句话说，如果记录执行失败，则不会提交具有包含这些记录的快照的检查点。因此，失败的检查点的记录将被完全重试。

这是可行的，因为 Cassandra 接收器有一个检查点提交程序，它在某些资源中存储有关已完成检查点的附加信息。此信息用于防止在发生故障时完全重播上次完成的检查点 [ref]。

【讨论】：