apache-flink - Apache Flink 检查点卡住

Question

我们正在运行一个 ListState 介于 300GB 到 400GB 和 之间的作业有时 该列表可以增长到几千个。在我们的用例中，每个项目都必须有自己的 TTL，因此我们为这个 ListState 的每个新项目创建一个新的 Timer，并在 S3 上使用 RocksDB 后端。

目前大约有 140 多个计时器(将在 event.timestamp + 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 检查点模式下，作业仍然正常运行和消耗。

这是我们第二次遇到拓扑运行良好，每天有几百万个事件，突然停止检查点。我们不知道是什么导致了这种情况。

任何人都可以想到什么会突然导致检查点卡住？

Answer 1

一些想法:

如果您有许多计时器或多或少同时触发，那么这场计时器 Storm 将阻止其他任何事情发生——任务将循环调用 onTimer 直到没有更多计时器被触发，在此期间它们的输入队列将被忽略，检查点障碍不会进展。

如果这是您遇到麻烦的原因，您可能会向计时器添加一些随机抖动，以便事件 Storm 以后不会变成计时器 Storm 。整理使用的东西 State TTL可能是另一种选择。

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

另外:由于您使用的是 RocksDB，从 ListState 切换到 MapState，以时间为键，可以让您删除单个条目，而无需在每次更新后重新序列化整个列表。 (对于 RocksDB，MapState 中的每个键/值对都是一个单独的 RocksDB 对象。)以这种方式提高清理效率可能是最好的补救措施。