使用检查点岩石SDB甩下过大负载

Question

目前，我们正试图找出如何有效地与Flink合作，我们仍在努力理解一切。

我们在一个独立的集群上运行大约60个非常轻量级的作业，这在普通的EC2实例上运行良好。然而，一旦我使用本地RocksDB状态后端启用检查点，集群的行为就会以意想不到的方式运行，停止作业，尝试重新启动它们，结果却丢弃了所有作业，并将错误日志留空。之后，Flink中没有留下作业或jar的任何痕迹。我知道对于每个作业，都会保留总JobManager内存的一小部分，同样，对于每个作业，都会在同一台机器上实例化一个本地RocksDB，但我认为它们同样轻量级，不需要太多内存/CPU容量。只需添加行env.enable检查点（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);

没有必要将类型转换为StateBindend，因为根据文档，RocksDBStateBindend类的1.9.1版本应该已经实现了StateBindend而不是AbstractStateBindend。然而，doc与我们从Maven获得的实际类并不相似，所以就是这样。

Answer 1

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

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

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

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