夜雨聆风基于 Spark 4.2，分支
branch-4.2阅读时长约 10 分钟 · 入门到中级
背景
Spark 官方文档说 RDD 有五个特性（Five Properties）, 面试也经常考。
这个文章做一件事：打开 RDD.scala，把五个特性逐个指到具体的代码行。
一、RDD 注释里的原文
打开 RDD.scala:71-75，Spark 自己写了五个特性：
* - A list of partitions
* - A function for computing each split
* - A list of dependencies on other RDDs
* - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
* - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
* an HDFS file)
对应的抽象类定义在 RDD.scala:84：
abstract class RDD[T: ClassTag](@transient private var sc: SparkContext,@transient private var deps: Seq[Dependency[_]]) extends Serializable with Logging
下面逐个拆开。
二、特性 1：一组 Partitions
含义：一个 RDD 由一组 Partition 组成，每个 Partition 是数据的逻辑分片。
源码体现：
RDD.scala:125：抽象方法getPartitionsprotected def getPartitions: Array[Partition]RDD.scala:296：final 方法partitions，带 checkpoint 逻辑final def partitions: Array[Partition] = {// checkpoint or lazily computegetPartitions}Partition.scala:23：Partition traittrait Partition extends Serializable {def index: Int}
Partition 本身不存数据。它只是一个标识——告诉 Spark "第 N 个分片在这里"。
子类实现举例：
ParallelCollectionRDD.scala:96：getPartitions把本地集合切成多个 sliceMapPartitionsRDD.scala:54：getPartitions直接返回父 RDD 的 partitions（窄转换不改分区）ShuffledRDD.scala:92：getPartitions根据partitioner.numPartitions生成新分区
三、特性 2：每个 Partition 的计算函数
含义：给定一个 Partition，RDD 知道怎么算出它的数据。
源码体现：
RDD.scala:116：抽象方法computedef compute(split: Partition, context: TaskContext): Iterator[T]RDD.scala:334：final 方法iterator，统一处理 cache/checkpointfinal def iterator(split: Partition, context: TaskContext): Iterator[T] = {if (storageLevel != StorageLevel.NONE) {getOrCompute(split, context)} else {computeOrReadCheckpoint(split, context)}}
iterator 是 Task 执行时实际调用的入口（见上一篇文章 ResultTask.runTask 里的 rdd.iterator(partition, context)）。它先查缓存，没有缓存才调 compute。
子类实现举例：
ParallelCollectionRDD.scala:101：compute直接从 Partition 的 iterator 读数据MapPartitionsRDD.scala:56：compute用函数f包装父 RDD 的 iteratoroverride def compute(split: Partition, context: TaskContext): Iterator[U] =f(context, split.index, firstParent[T].iterator(split, context))ShuffledRDD.scala:102：compute通过 ShuffleManager reader 读取 shuffle 输出
四、特性 3：对其他 RDD 的依赖列表
含义：RDD 知道自己依赖谁。这个依赖列表构成了 RDD 血缘（lineage），也是容错和 Stage 划分的基础。
源码体现：
RDD.scala:131：getDependencies，默认返回构造参数depsprotected def getDependencies: Seq[Dependency[_]] = depsRDD.scala:260：final 方法dependencies，带 checkpoint 逻辑final def dependencies: Seq[Dependency[_]] = {// checkpoint or lazily computegetDependencies}Dependency.scala:41：Dependency抽象类abstract classDependency[T] extendsSerializable{def rdd: RDD[T]}
每个 Dependency 对象指向一个父 RDD，就是血缘图中的一条边。
窄依赖和宽依赖：
Dependency.scala:52：NarrowDependency，子 partition 只依赖少量父 partitionDependency.scala:58：getParents(partitionId)描述子到父的映射Dependency.scala:84：ShuffleDependency，宽依赖，需要 shuffleDependency.scala:266：OneToOneDependency，最典型的窄依赖
子类实现举例：
MapPartitionsRDD.scala:50：extends RDD[U](prev)—— 用单父构造器，自动产生OneToOneDependency（RDD.scala:104）ShuffledRDD.scala:78：getDependencies返回List(new ShuffleDependency(...))ParallelCollectionRDD.scala:90：extends RDD[T](sc, Nil)—— 没有父依赖，deps = Nil
五、特性 4：可选的 Partitioner
含义：对于 key-value 类型的 RDD，可以声明分区策略（hash / range 等）。这影响 shuffle 输出的分区方式，也影响 join 等操作的优化。
源码体现：
RDD.scala:139：partitioner字段@transient val partitioner: Option[Partitioner] = None
大多数 RDD 的 partitioner 是 None。只有经过 shuffle 或显式 repartition 的 RDD 才有。
子类实现举例：
ShuffledRDD.scala:90：override val partitioner = Some(part)MapPartitionsRDD.scala:52：如果preservesPartitioning为 true，继承父 partitioneroverride val partitioner = if (preservesPartitioning) firstParent[T].partitioner else None
ShuffleDependency 也持有 partitioner（Dependency.scala:86），保证 shuffle 读写双方的分区策略一致。
六、特性 5：可选的 Preferred Locations
含义：RDD 可以建议"某个 Partition 最好在哪个机器上算"。这是数据本地性的基础。
源码体现：
RDD.scala:136：getPreferredLocations，默认返回空protected def getPreferredLocations(split: Partition): Seq[String] = NilRDD.scala:323：final 方法preferredLocationsfinal def preferredLocations(split: Partition): Seq[String] = {getPreferredLocations(split)}
子类实现举例：
ShuffledRDD.scala:96：getPreferredLocations通过MapOutputTracker查询 shuffle 数据块的位置override def getPreferredLocations(partition: Partition): Seq[String] = {val tracker = SparkEnv.get.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]tracker.getPreferredLocationsForShuffle(part, partition.index)}ParallelCollectionRDD.scala:105：从locationPrefs读取位置偏好
数据本地性的调度策略在 TaskSetManager 里使用：优先分配 Task 到 PROCESS_LOCAL，其次 NODE_LOCAL，依次降级到 ANY。
七、五大特性一览表
| # | 特性 | RDD.scala 位置 | 抽象/默认 | 子类典型实现 |
|---|---|---|---|---|
| 1 | Partitions | getPartitions:125 / partitions:296 | 抽象 | ParallelCollectionRDD 切片；MapPartitionsRDD 继承父 |
| 2 | Compute | compute:116 / iterator:334 | 抽象 | MapPartitionsRDD 包装函数；ShuffledRDD 读 shuffle |
| 3 | Dependencies | getDependencies:131 / dependencies:260 | 默认返回构造参数 | ShuffledRDD 返回 ShuffleDependency |
| 4 | Partitioner | partitioner:139 | 默认 None | ShuffledRDD 声明 partitioner |
| 5 | Preferred Locations | getPreferredLocations:136 / preferredLocations:323 | 默认 Nil | ShuffledRDD 查 MapOutputTracker |
八、三个子类对比
| 子类 | Partitions | Compute | Dependencies | Partitioner | Preferred Locations |
|---|---|---|---|---|---|
| ParallelCollectionRDD | 切本地集合 (:96) | 从 Partition 读 (:101) | Nil（无父 RDD）(:90) | None | locationPrefs (:105) |
| MapPartitionsRDD | 继承父 (:54) | f(父.iterator) (:56) | OneToOneDependency（自动）(:50) | 继承父或 None (:52) | 默认 Nil |
| ShuffledRDD | partitioner 决定 (:92) | 读 shuffle (:102) | ShuffleDependency (:78) | 声明 partitioner (:90) | MapOutputTracker (:96) |
这三个子类几乎覆盖了所有场景：
源头 RDD（如 ParallelCollectionRDD / HadoopRDD）：自己造分区、自己算数据
窄转换 RDD（如 MapPartitionsRDD）：直接继承分区和依赖，只在 compute 里加一层函数
宽转换 RDD（如 ShuffledRDD）：创建新的分区、新的依赖、新的 partitioner
九、实战：用源码验证五大特性
object TestRDD {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("RDD-DEMO").master("local[4]").config("spark.serializer", "org.apache.spark.serializer.JavaSerializer").getOrCreate();val sc = spark.sparkContextsc.setLogLevel("WARN")val rdd1 = sc.parallelize(1 to 100, 4)val rdd2 = rdd1.map(_ * 2)val pairs = sc.parallelize(Seq(("a", 1), ("b", 2), ("a", 3)), 2)val rdd3 = pairs.reduceByKey(_ + _)// 特性 1: partitionsprintln(s"rdd1 partitions: ${rdd1.getNumPartitions}")println(s"rdd3 partitions: ${rdd3.getNumPartitions}")// 特性 3: dependenciesprintln(s"rdd1 deps: ${rdd1.dependencies.map(_.getClass.getSimpleName)}")println(s"rdd2 deps: ${rdd2.dependencies.map(_.getClass.getSimpleName)}")println(s"rdd3 deps: ${rdd3.dependencies.map(_.getClass.getSimpleName)}")// 特性 4: partitionerprintln(s"rdd1 partitioner: ${rdd1.partitioner}")println(s"rdd3 partitioner: ${rdd3.partitioner}")// 特性 5: preferredLocationsprintln(s"rdd1 preferred: ${rdd1.preferredLocations(rdd1.partitions(0))}")}}
输出：
rdd1 partitions: 4
rdd3 partitions: 2
rdd1 deps: List()
rdd2 deps: List(OneToOneDependency)
rdd3 deps: List(ShuffleDependency)
rdd1 partitioner: None
rdd3 partitioner: Some(org.apache.spark.HashPartitioner@2)
rdd1 preferred: List()
十、总结
RDD 的五大特性不是面试八股，而是 Spark 调度系统的地基：
Partitions 决定了并行度
Compute 决定了每个 Task 做什么
Dependencies 决定了 Stage 怎么切、数据怎么传
Partitioner 决定了 shuffle 输出怎么分
Preferred Locations 决定了 Task 往哪台机器放
从源码角度看，RDD 的设计非常统一：五个特性对应五个可覆盖的方法/字段，子类只需选择性地覆盖自己关心的部分。
每天花费5分钟学习spark，让你技术之路走得更稳、更快。
喜欢的点个关注。
