使用 Checkpointing RocksDB 进行 Flink 负载过大答案

【问题标题】：Flink excessive load with Checkpointing RocksDB使用 Checkpointing RocksDB 进行 Flink 负载过大
【发布时间】：2020-03-10 09:20:34
【问题描述】：

目前，我们正在尝试弄清楚如何有效地使用 Flink，并且我们仍在尝试理解一切。

我们在一个独立集群上运行了大约 60 个真正轻量级的作业，这在一个普通的 EC2 实例上运行良好。但是，一旦我使用本地 RocksDB 状态后端启用检查点，集群就会以意想不到的方式运行，停止作业，尝试重新启动它们只是丢弃所有它们并将错误日志留空。之后，在 Flink 中没有留下任何作业或 jar 的痕迹。我知道，对于每个作业，都会保留 JobManager 总内存的一小部分，同样，对于每个作业，本地 RocksDB 都会在同一台机器上实例化，但我认为它们同样轻巧，不需要太多内存/CPU 容量.与之前的稳定集群相比，只需添加行 env.enableCheckpointing(1000); 就会导致一切完全失败。

我个人认为我们可能已经达到了我们独立 Flink 集群的极限，即使增加内存也不够了，但我想在开始构建分布式 Flink 集群之前确认这一点（我们需要自动化一切，这就是我现在犹豫的原因）。我不确定是否例如将 RocksDB 检查点存储在像 S3 这样的专用存储单元中甚至可以解决这个问题，如果资源消耗（硬盘除外）会受到影响。

迁移到分布式环境是解决我们问题的唯一方法，还是这表明存在其他问题，可以通过适当的配置来解决？

编辑：也许我应该补充一点，还没有加载，我们还没有谈论传入的数据，但关于作业仍在运行。目前 FlinkSources 中只有 100 条记录，但我们甚至无法达到正在处理的程度。

编辑2：

这部分一直是作业代码的一部分：

try {
    env.setStateBackend((StateBackend) new RocksDBStateBackend("file://" + "/somePathOnInstance"));
    } catch (IOException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    }

我们添加了以下代码行：

env.enableCheckpointing(CHECKPOINTING_INTERVAL_MS);

不需要对 StateBackend 进行类型转换，因为根据文档，RocksDBStateBackend 类的 1.9.1 版本应该已经实现了 StateBackend 而不是 AbstractStateBackend。但是，该文档与我们从 Maven 获得的实际类不同，所以就是这样。

【问题讨论】：

您是否同时启用了检查点并切换到使用 RocksDB，或者您之前是否成功使用过 RocksDB，但没有检查点？
@DavidAnderson 我们之前已经启用了 RocksDB，我在最初的帖子中添加了详细信息，请查看最近的编辑。但是，如果不使用检查点，RocksDB 可能只是用于捕获状态，在我们的例子中它非常小（部分只是一个布尔值）。我只能假设它比使用实际检查点占用更少的资源？
开启检查点确实会增加资源需求。您真的在单个 EC2 实例上同时运行 60 个作业吗？你同时为所有这些都打开了检查点？
@DavidAnderson 从这个问题来看，我认为这很不寻常？答案是肯定的，60 个工作（主要是 1 个源、1 个接收器和一个介于两者之间的简单函数），这实际上在 t3 上没什么大问题。中等实例没有检查点。然而，没有繁重的负载，最多只有几千条记录，但它是这样工作的。
我不知道这是否会有所改进，但我只是指出可以在同一个作业中运行多个管道。使用这种技术来减少工作的总数应该会减少总体资源需求。

标签： apache-flink

【解决方案1】：

鉴于您正在运行的 60 个作业的工作负载相当微不足道，因此有理由认为启用检查点会产生重大影响。基本上，我怀疑拥有 60-180 个新线程（我不确定您的哪些操作员是有状态的）都试图频繁写入文件系统会压倒您的 t3.medium 实例。

检查点由检查点协调器（在 Flink 主控器中）管理，它与所有作业通信、启动检查点、等待它们完成并管理元数据。所涉及的大部分工作都是由任务管理器完成的，并且是异步完成的，所以在你的情况下，有很多新线程，每个线程都将被检查点的数据复制到你的检查点存储（这应该是一个分布式文件系统，比如S3)。如果检查点间隔很短，例如一秒，那么这一切都是每秒发生的。

您可以检查各种指标以试图找出瓶颈所在——您可能受到内存、CPU 或网络带宽的限制。

使用 RocksDB，增量检查点通常比完整检查点更轻，因此选择该选项会有所帮助 - 尽管状态量如此之少，但我不认为这会有所帮助。更重要的是，出于性能原因，您应该避免使用 EBS（或任何其他网络附加存储）作为 RocksDB 的本地磁盘。

【讨论】：

感谢详细解答，我会在此基础上继续调查。我最初的问题出在其他地方，但我设法解决了它，现在我收到错误日志消息，这很清楚地表明内存不足。现在，作业是高度模块化的（因此数量很大），但实际上每 4-5 个作业都是链式的，即一个流的输出是另一个流的输入。在资源方面将这些整合到一份工作中是否有显着优势？我们喜欢这里的逻辑分离，但还不知道我们会在多大程度上损害效率。
不惊讶回复：内存。合并可能会有所帮助，但如果你想确定的话，你应该做一些测试。在更大的实例上运行可能是一个更快乐的解决方案；很难说。
从长远来看，除了迁移到分布式集群之外别无他法，增加机器大小并不是一种令人满意的可扩展性形式。我们将很快解决这个问题，并希望获得一些新的见解。
扩展到分布式集群应该很容易。如果您觉得它具有挑战性，请提出一两个新问题，我们会尽力提供帮助。