我有一个Flink工作处理数据在约20万qps。没有检查点，作业运行正常。但是，当我尝试添加检查点（间隔50分钟）时，它会导致在第一个任务中出现backpressue，即为每个条目添加一个键字段，数据延迟也会不断增加。我的两个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分钟的检查站看看
我试图调整内存使用，但似乎不起作用。
但在任务管理器中，仍然是：
2条答案
按热度按时间bn31dyow1#
tldr公司；有时很难分析这个问题。我有两个幸运的猜测/镜头-如果你正在使用rocksdb状态后端，你可以切换到fsstatebend-它通常更快，rocksdb最有意义的大状态大小，不适合内存（或如果你真的需要增量检查点功能）。第二是摆弄平行度，不是增加就是减少。
我会怀疑@arvidheise写的同样的东西。检查点的大小不是很大，但也不是很小。它会增加额外的开销，使作业超过勉强跟上流量的阈值，从而导致无法跟上并产生背压。如果您处于背压之下，延迟将不断累积，因此即使额外开销的两个百分点发生变化，也会使端到端延迟（毫秒）与无限增长值之间产生差异。
如果您不能简单地添加更多的资源，那么您必须分析到底是什么在添加额外的资源，以及什么资源是瓶颈。
是cpu吗？检查群集上的cpu使用情况。如果它是~100%，那就是你需要优化的东西。
是伊奥吗？检查集群上的io使用情况，并将其与您可以实现的最大吞吐量/每秒请求数进行比较。
如果cpu和io使用率都很低，您可能希望尝试增加并行性，但是。。。
请记住数据倾斜。背压可能是由单个任务引起的，在这种情况下，很难分析问题，因为它将是单个瓶颈线程（在io或cpu上），而不是整个机器。
在弄清楚什么资源是瓶颈之后，下一个问题是为什么？一旦您看到它，它可能会立即变得很明显，或者可能需要深入研究，比如检查gc日志、附加profiler等等。
回答这些问题可以为您提供信息，您可以尝试优化您的工作，或允许您调整配置，或可以给我们（flink开发人员）一个额外的数据点，我们可以尝试优化flink方面。
sf6xfgos2#
任何检查点都会增加计算开销。大多数检查点都是异步的（如您所述），但它仍然会增加一般的i/o操作。例如，这些额外的i/o请求可能会阻塞您对外部系统的访问。另外，如果启用了检查点，flink需要跟踪更多的信息（新的和已经检查过的）。
你试过给你的工作增加更多的资源吗？你能分享你的整个检查点配置吗？