scala - 为什么使用 updateStateByKey 时任务大小一直在增长？

Question

我编写了一个与 updateStateByKey 一起使用的简单函数，以查看问题是否是因为我的 updateFunc。我认为这一定是由于其他原因。我在 --master local[4] 上运行它。

val updateFunc = (values: Seq[Int], state: Option[Int]) => {
  Some(1)
}

val state = test.updateStateByKey[Int](updateFunc)

一段时间后，有警告，任务大小不断增加。

WARN TaskSetManager:阶段 x 包含一个非常大的任务 (129 KB)。建议的最大任务大小为 100 KB。

WARN TaskSetManager:阶段 x 包含一个非常大的任务 (131 KB)。建议的最大任务大小为 100 KB。

Answer 1

您的流中有越来越多的不同键，每个键都会将 1 的新副本添加到您的状态。

当前 updateStateByKey 在每个批处理间隔中扫描每个键，即使该键没有数据。这会导致 updateStateByKey 的批处理时间随着状态中键的数量而增加，即使数据速率保持固定。

有一个proposal to solve this .