Flink检查点随机失败

Question

我正在kubernetes中运行一个Flink作业。我的设置如下

1. 1个作业管理器吊舱

我的flink作业从kafka源获取时间序列数据（时间、值），进行聚合和其他转换，并将其发布到kafka接收器。

有时，我的作业因检查点异常（10分钟后超时）而失败，主要是由一名操作员完成的。我不理解异步持续时间（在图中）的含义，为什么它花费的时间最长。在这个异常之前，Kafka的吞吐量非常高，有500-800万条记录，但在这个检查点异常之后，吞吐量变得非常慢。另一个观察结果是，增加并行性无助于提高吞吐量。我还启用了未对齐的检查点，但它没有多大帮助。

Flink版本：1.13.2提交：5f007ff@2021-07-23T04:35:55 02:00

感谢您的帮助。

编辑：

在David的评论之后，我对我的集群进行了以下更改

我添加了一个单独的节点池，每个节点附有10个节点和1个本地ssd（默认大小为375 GB），并在此节点池中调度我的TM pod。此外，我将rocksdb localdir设置为溢出状态数据到磁盘。我有1个kafka源，可以流式传输10亿时间序列数据

检查点间隔=15分钟，还启用了未对齐的检查点

1. 第一次运行，吞吐量非常高。它处理

问题

1. 为什么第一次运行时吞吐量真的很高，然后又下降了
2. 首次运行时仍观察到检查点故障
3. 在首次运行时取消作业并重新提交作业无助于获得相同的吞吐量
4. 虽然作业现在运行良好，没有检查点故障，但为什么所有本地ssd设置的吞吐量仍然很低
5. 为什么检查点大小波动如此之大？有时多，有时少

Answer 1

看起来您有一个用例，其中未对齐的检查点实际上没有任何帮助，反而使其恶化。

未对齐的检查点适用于遇到突然峰值或持续时间较长的背压的用例，因此在开始时吞吐量非常低的情况下。

现在，未对齐的检查点正在检查点中持久化瞬态数据。这就是为什么大小会膨胀，实际上可能会有很大的变化，这取决于网络缓冲区饱和的程度。

从未对齐的检查点恢复也需要更长的时间（要恢复更多数据）。

现在的问题仍然是，为什么对未对齐的检查点会有超时。理论上，只有当您的I/O成为瓶颈时，才会发生这种情况。未对齐的检查点将您的检查点大小从1 GB放大到9 GB。

另一种解释可能实际上是水槽侧的一个完全瓶颈；由于您可能使用了KafkaConsumer，而不是新的KafkaSource，因此需要最少的数据流。但我想从你的解释中排除这一点。

为了进一步调查，我需要一个超时检查点的屏幕截图（每个任务的详细统计数据）和日志。

最后一个问题是，即使作业完全重新启动，总体吞吐量仍然存在问题。根据提供的信息，这个问题很难回答。我猜您有一些有状态操作，其中heap statebend在达到某个状态大小后运行得不太理想。您可以尝试调整StateBend。例如，确保调整SSD的rocks DB。

Answer 2

完全同意David的意见，您应该使用本地磁盘。这里没有耐久性，只有性能要求。有关详细信息，请查看此博文[1]。至于吞吐量下降，您是否检查了本地磁盘的指标？它们是不稳定的磁盘吗？你可能会遇到瓶颈。查看这篇博文[2]，看看它是否符合您的情况。如果不是磁盘，也检查CPU

[1]https://flink.apache.org/2021/01/18/rocksdb.html

[2]https://www.ververica.com/blog/the-impact-of-disks-on-rocksdb-state-backend-in-flink-a-case-study

Answer 3

关于您的设置，我要更改的第一件事是移动状态。后端。rocksdb。localdir到每个TM pod的本地临时存储。如果RocksDB本地磁盘实际上是NFS装载，则可能会导致问题。

至于异步持续时间，检查点分两个阶段完成。一个（希望简短）同步阶段，在此期间流处理暂停，以便可以检查点一些关键内容，然后是一个通常更长的异步阶段，在此期间键控状态被检查点（在与正在进行的流处理分开的线程中）。