在持久化 RDD 上有多个操作的情况下，缓存 RDD 的工作原理答案

【问题标题】：How Caching an RDD works in case of multiple actions on a persisted RDD在持久化 RDD 上有多个操作的情况下，缓存 RDD 的工作原理
【发布时间】：2019-01-08 05:48:20
【问题描述】：

val logList: RDD[String] = ...
val errorLogs = logList.filter(_.contains("Error")).persist()
//first action   
val first100 = errorLogs.take(100)
//second action
val count = errorLogs.count

persist 在这种情况下将如何工作？在以下代码的情况下

val errorLogs = logList.filter(_.contains("Error")).take(100)

Spark 不会扫描所有日志，因为Spark 知道我们只对 100 行日志感兴趣。但是当我们缓存这个 RDD 并对其调用多个操作时会发生什么，第一个操作只需要很少的记录，而后面的操作需要转换整个 RDD。

当第一个动作被调用时它会缓存记录吗？还是仅缓存调用第一个操作时第一个操作所需的部分记录？

【问题讨论】：

标签： scala apache-spark caching rdd

【解决方案1】：

在这种情况下，Spark 将只缓存收集 100 条记录所需的最小分区数（由于采取实施，实际数量可以更高）。

只有第二个保证缓存所有记录。

【讨论】：

看起来很合理，但是。因此，以第一种情况为例，让我与您核对一下：假设日志文件根据磁盘大小有 22 个分区，那么，如果所有错误都驻留在 12 个分区上 --> 并且包含 2.2M 条记录，那么什么是考虑到 RDD 并行处理的一般原则和并行发生的事情，收集 100 条记录所需的最少记录数？如果我们明白这一点，那么我们就知道答案，而且很可能你知道吗？因此？也就是说，我发现这方面的文档很难找到。
只有第二个保证缓存所有记录...你的意思是 val count = errorLogs.count OR val errorLogs = logList.filter(_.contains("Error")).take(100 ）。不知道我明白了。
这很难，恕我直言。问题是什么是持久化的 RDD。
@thebluephantom：第二个意思是，调用操作errorLogs.count。因为 count 必须扫描整个数据集才能得到实际结果。所以在这个动作期间，errorLogs 会被完全缓存。
这似乎是有道理的，但我的第一个问题可能意味着实际上所有内容都被缓存或超过最小值被缓存，原因是 1) 无论如何并行化和 2) 如果没有找到记录，它将全部缓存为它必须扫描所有内容才能决定

【解决方案2】：

cache 的工作原理：

cache 和 persist 是惰性的 - 如果没有对标记为缓存数据的 RDD 调用任何操作，则不会持久保存。同样，如果未评估分区，则数据不会持久化。
cache 保留整个分区。它不能保留单个记录或部分分区。如果数据被标记为缓存，并且至少部分评估了分区，Spark 将对其进行全面评估并尝试持久化。

limit 的工作原理：

First limit 评估第一个分区。如果可以从第一个分区收集限制所需的所有记录，则作业完成。
如果不是，Spark 会增加要由spark.rdd.limit.scaleUpFactor 评估的分区数。如果可以从这些分区收集限制所需的所有记录，则作业完成。否则重复此步骤。

这意味着：

val first100 = errorLogs.take(100) 将至少缓存第一个分区。如果第一个分区不包含 100 条记录，它将评估并缓存后续分区，直到达到 100 条记录或评估完整的数据集。
val count = errorLogs.count 将评估所有分区并尽可能缓存剩余的分区。
val errorLogs = logList.filter(_.contains("Error")).take(100) 的工作方式与errorLogs.take(100) 几乎完全相同。 filter 的唯一影响是 limit 可能需要评估更多数据。如果所有行都包含Error，则结果与第一步相同。

如果这在前两个步骤之后运行，并且数据已完全缓存并且尚未被驱逐，它将使用缓存中的数据。

【讨论】：