Spark中的RDD依赖是啥？

Posted 2023-04-18

【中文标题】Spark中的RDD依赖是啥？

【英文标题】：What is RDD dependency in Spark?Spark中的RDD依赖是什么？

【发布时间】：2015-02-14 10:11:16

【问题描述】：

据我所知，有两种依赖类型：窄和宽。但我不明白依赖性如何影响child RDD。 child RDD 是否只有元数据，其中包含如何从 parent RDD 构建新 RDD 块的信息？或者 child RDD 是从 parent RDD 创建的自给自足数据集？

【参考方案1】：

是的，子 RDD 是描述如何从父 RDD 计算 RDD 的元数据。

以org/apache/spark/rdd/MappedRDD.scala 为例：

private[spark]
class MappedRDD[U: ClassTag, T: ClassTag](prev: RDD[T], f: T => U)
  extends RDD[U](prev) 

  override def getPartitions: Array[Partition] = firstParent[T].partitions

  override def compute(split: Partition, context: TaskContext) =
    firstParent[T].iterator(split, context).map(f)

当您说rdd2 = rdd1.map(...) 时，rdd2 将是这样的MappedRDD。 compute 仅在稍后执行，例如当您调用 rdd2.collect 时。

RDD 总是这样的元数据，即使它没有父级（例如sc.textFile(...)）。 RDD 存储在节点上的唯一情况是使用rdd.cache 将其标记为缓存，然后计算它。

另一个类似的情况是调用rdd.checkpoint。该函数将 RDD 标记为检查点。下次计算时，它会被写入磁盘，以后访问 RDD 会导致它从磁盘读取，而不是重新计算。

cache 和 checkpoint 的区别在于缓存的 RDD 仍然保留其依赖关系。缓存的数据在内存压力下可能会被丢弃，可能需要部分或全部重新计算。带有检查点的 RDD 不会发生这种情况，因此依赖项会在那里被丢弃。

【讨论】：

感谢您的明确答复。但是为了澄清 - 据我了解，新转换的 RDD 只是指向某些（过滤的）旧数据块的新指针集？还是新转换的 RDD 是旧数据的新副本？我对在转换过程中与 RDD 集群发生的物理事物感兴趣。

RDD 是惰性的。在执行急切的操作（如collect 或reduce）之前，不会执行任何工作。当一个动作最终被执行时，像map和filter这样的操作作为一个迭代器链执行。重要的一点是，RDD 通常不代表数据，它代表计算。

好的，但是让我们想象一下我们有 Spark 作业，并进行了下一步的计算：(1)RDD -> (2)map->(3)filter->(4)collect。在第一阶段我们输入 RDD，在第二阶段我们将这些 RDD 转换为 map(kay-value 对)。那么Spark在第三阶段过滤的结果是什么呢？ Spark 会从 RDD 中删除不必要的项目吗？或者它会用必要的项目创建绝对新的 RDD 并删除以前的 RDD？过滤后不需要的父 RDD 的一堆项目会怎样？

RDD 是用迭代器实现的。因此将读取输入文件，并逐行应用 map 函数，然后是 filter 函数。永远不会存储超过一行。 （好吧，它们会在内存中徘徊，直到垃圾收集器清理它们。）collect 是个例外，它调用 iterator.toArray 将结果转换为数组并将它们发送回应用程序。