Flink FsStateBackend 使用 S3 存储太贵了

Question

目前我使用 FsStateBackend 检查点状态。我正在使用间隔 10 秒，如下面的代码。但我看到使用检查点的传输桶的成本约为 20 美元/天，aws transfer s3 定价：0.005 美元/1000 个请求 =>（我使用的是 ~4000000 个请求/天 @@）。我有 7 个工作，其中：

• 6 个作业使用检查点间隔 = 10000 (ms)
• 1 个作业使用检查点间隔 = 1000 (ms)

并在 AWS EMR 上运行 flink。每个检查点的平均状态大小（8KB -> 30M）。检查点后面发生了什么？

// set up checkpoint
        env.enableCheckpointing(1000 or 10000);

        // advanced options:
        // make sure 500 ms of progress happen between checkpoints
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // checkpoints have to complete within one minute, or are discarded
//            env.getCheckpointConfig().setCheckpointTimeout(60000);
        // allow only one checkpoint to be in progress at the same time
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // enable externalized checkpoints which are retained after job cancellation
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // folder to checkpoint
        StateBackend backend = new FsStateBackend(checkpointPath, true);
        env.setStateBackend(backend);

Answer 1

您将哪个 S3 实现用于检查点？这有很大的不同。

虽然您必须将 S3 的 hadoop 实现与 StreamingFileSink 一起使用，但它是检查点的一个非常糟糕的选择。 Hadoop S3 FS 尝试在 S3 之上模仿文件系统：

• 在写入键之前，它会通过检查前缀直到最后一个“/”的键来检查“父目录”是否存在
• 它会创建空的标记文件来标记这样的父目录的存在
• 所有这些“存在”请求都是昂贵的 S3 HEAD 请求

因此，Hadoop S3 FS 的“创建文件”延迟非常高，并且会很快达到请求速率限制（HEAD 请求在 S3 上的请求速率限制非常低）。

Presto S3 并没有尝试做到这一点。 它只是简单地执行 PUT/GET 操作，而没有围绕它的所有其他东西。 因为 Flink 的 checkpointing 假设仅此而已，所以它更加高效和一致。

此外，使用 Hadoop S3，您可能会遇到恢复操作失败的情况，因为看起来状态文件不存在（HEAD 请求导致 S3 负载均衡器中的错误缓存）。过了一会儿文件才可见，然后恢复成功。

请注意，可以将 hadoop 版本用于接收器，也可以将 presto 版本用于检查点。在这种情况下，您应该明确使用 s3a:// 作为接收器 (Hadoop) 的方案，并使用 s3p:// 作为检查点 (Presto)。

Flink S3 docs.