Apache Flink 检查点卡住。

Question

Apache Flink 检查点卡住。

4

我们正在运行一个作业，其中包含300GB到400GB的ListState，有时列表可以增长到数千个。在我们的用例中，每个项目必须具有自己的TTL，因此我们为此ListState的每个新项目创建一个新的定时器，并使用S3上的RocksDB后端。

目前大约有1.4亿个计时器（将在事件时间戳+40天触发）。

我们的问题是，突然作业的检查点会卡住，或者非常缓慢（比如几个小时内只有1%），直到最终超时。它通常会停止（flink仪表板显示0/12 (0%)，而之前的行显示12/12 (100%)），原因是代码片段非常简单：

[...]
    val myStream = env.addSource(someKafkaConsumer)
      .rebalance
      .map(new CounterMapFunction[ControlGroup]("source.kafkaconsumer"))
      .uid("src_kafka_stream")
      .name("some_name")

      myStream.process(new MonitoringProcessFunction()).uid("monitoring_uuid").name(monitoring_name)
        .getSideOutput(outputTag)
        .keyBy(_.name)
        .addSink(sink)
[...]

更多信息如下：

AT_LEAST_ONCE检查点模式似乎比EXACTLY_ONCE更容易卡住。
几个月前，状态达到了1.5TB的数据和数十亿的计时器，没有任何问题。
运行两个任务管理器的机器上的RAM、CPU和网络看起来正常。
state.backend.rocksdb.thread.num = 4
第一次事故发生在我们收到大量事件（每分钟约有数百万个）时，但上一次没有发生这种情况。
所有事件都来自Kafka主题。
当处于AT_LEAST_ONCE检查点模式时，作业仍然正常运行和消耗。

这已经是我们第二次遇到拓扑结构运行正常，每天只有几百万个事件，然后突然停止检查点。我们不知道是什么原因导致的。

有人能想到是什么原因突然导致检查点卡住吗？

- Rocel

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- David Anderson · Accepted Answer

几点想法：

如果您有许多计时器几乎同时触发，这些计时器的风暴将阻止其他任何事情发生 - 任务将循环调用 onTimer，直到没有更多计时器需要触发，在此期间它们的输入队列将被忽略，并且检查点障碍不会进展。

如果这是您遇到问题的原因，您可以向计时器添加一些随机抖动，以便事件风暴不会在以后变成计时器风暴。重新组织使用状态TTL可能是另一个选择。

如果堆上有大量计时器，这可能导致非常高的GC开销。这并不一定会使作业失败，但可能会使检查点不稳定。在这种情况下，将计时器移入RocksDB可能会有所帮助。

还有：由于您正在使用RocksDB，从ListState切换到MapState，以时间为键，可以让您在每次更新后删除单个条目而无需重新序列化整个列表。（对于RocksDB中的每个MapState中的每个键/值对都是单独的RocksDB对象。）以这种方式使清理更有效可能是最好的解决方法。