Flink 检查点导致背压

Question

我有一个大约 200k qps 的 Flink 作业处理数据。没有检查点，作业运行良好。 但是当我尝试添加检查点（间隔 50 分钟）时，它会在第一个任务中导致背压，即为每个条目添加一个关键字段，数据滞后也不断上升。 我的两个 Kafka 主题的滞后，上半年启用了检查点，滞后很快上升。第二部分（非常低的滞后禁用了检查点，滞后在毫秒内）

我至少在使用once checkpoint mode，它应该是异步进程。有人可以建议吗？ 我的检查点设置

    env.enableCheckpointing(1800000,
          CheckpointingMode.AT_LEAST_ONCE);
      env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
      env.getCheckpointConfig()
          .enableExternalizedCheckpoints(
              CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
      env.getCheckpointConfig()
          .setCheckpointTimeout(10min);
      env.getCheckpointConfig()
          .setFailOnCheckpointingErrors(
              jobConfiguration.getCheckpointConfig().getFailOnCheckpointingErrors());

我的工作有 128 个容器。

检查点时间为 10 分钟，以下是统计数据：

我正在尝试使用 30 分钟检查站并查看

我试图调整内存使用，但它似乎不起作用。

但在任务管理器中，它仍然是：

Answer 1

TLDR;有时很难分析问题。我有两个幸运的猜测/镜头 - 如果你使用 RocksDB 状态后端，你可以切换到 FsStateBackend - 它通常更快，而且 RocksDB 最适合大型状态，不适合内存（或者如果你真的需要增量检查点功能）。其次是摆弄并行度，增加或减少。

我会怀疑@ArvidHeise 写的同样的事情。你的检查点大小不是很大，但也不是微不足道的。它可以增加额外的开销，使工作超过勉强跟上流量的阈值，不跟上并导致背压。如果您处于背压之下，则延迟会不断累积，因此即使是几 % 的额外开销的变化也会在毫秒级的端到端延迟与无限增长的值之间产生差异。

如果您不能简单地添加更多资源，则必须分析究竟是什么增加了这些额外开销以及什么资源是瓶颈。

1. 是 CPU 吗？检查集群上的 CPU 使用率。如果它是~100%，那就是你需要优化的东西。
2. 是 IO 吗？检查集群上的 IO 使用情况，并将其与您可以实现的最大吞吐量/每秒请求数进行比较。
3. 如果 CPU 和 IO 使用率都很低，您可能想尝试增加并行度，但是...
4. 注意数据倾斜。背压可能是由单个任务引起的，在这种情况下，很难分析问题，因为它将是单个瓶颈线程（在 IO 或 CPU 上），而不是整个机器。

弄清楚瓶颈是什么资源后，下一个问题是为什么？一旦你看到它可能会立即显而易见，或者它可能需要深入研究，比如检查 GC 日志、附加分析器等。

回答这些问题既可以为您提供信息，您可以在工作中尝试优化哪些内容，也可以让您调整配置，或者可以为我们（Flink 开发人员）提供额外的数据点，我们可以尝试在 Flink 方面进行优化。

Answer 2

任何类型的检查点都会增加计算开销。大多数检查点是异步的（如您所述），但它仍然会增加一般的 I/O 操作。例如，这些额外的 I/O 请求可能会阻塞您对外部系统的访问。此外，如果您启用检查点，Flink 需要跟踪更多信息（新的与已设置的检查点）。

您是否尝试为您的工作添加更多资源？你能分享你的整个检查点配置吗？