为什么 Spark 中有这么多 MapWithStateRDD答案

【问题标题】：Why are there so many MapWithStateRDDs in Spark为什么 Spark 中有这么多 MapWithStateRDD
【发布时间】：2018-10-23 05:02:26
【问题描述】：

我有一个使用状态转换的 Spark 流应用程序。回想起来，Spark 可能不是最佳选择，但我仍在努力让它发挥作用。

我的问题是，为什么我的 MapWithStateRDD 会占用这么多内存？例如，我有一个转换，其中状态在内存中约为 1.5 Gb，我看到每个批次都恢复了相同的 RDD。因此，在第三批之后，它在 UI 上显示有 3 个大小完全相同的 MapWithStateRDD，而这些批次中的状态没有改变。这些实际上占用了 3 倍的空间吗？这似乎是一种巨大的浪费，它不应该只将增量存储到一个检查点，然后将它们压缩到一个 RDD 或类似的东西上吗？我认为这就是它的工作原理，并且有更多的有状态转换会占用大量内存。

【问题讨论】：

Spark Streaming mapWithState seems to rebuild complete state periodically的可能重复
我不觉得这是重复的。我知道检查点间隔以及为什么这些批次需要更长的时间。我的问题是 UI 是否报告实际使用的内存，如果是，为什么即使状态没有改变，或者只有一小部分改变，它也会缓存每批中的状态。

标签： apache-spark spark-streaming

【解决方案1】：

正如 cmets 中的链接所指出的，这是因为 mapWithState 在每 10 个批次后检查一次数据，因此它会一直缓存 RDD 直到该点。

要消除这种空间浪费，您可以选择在每批检查点状态。就我而言，这并没有变得非常昂贵。

JavaInputDStream<ConsumerRecord<String, Object>> rtStream = ...
JavaMapWithStateDStream<String, Object, Object, Tuple2<String, Object>> mapWithStateSTream = rtStream .mapToPair(..).mapWithState(...);

mapWithStateSTream.checkpoint(Durations.seconds(10)); //keep this same as your batch interval, or feel free to change acc to your requirement

mapWithStateSTream.forEachRdd(...your logic here..);

你去吧，魔法！！您不再在“存储”选项卡中看到那些恼人的 MapWithStateRDD。

【讨论】：