RDD 的依赖关系

scala> var rdd1 = sc.textFile("./words.txt")rdd1: org.apache.spark.rdd.RDD[String] = ./words.txt MapPartitionsRDD[1] at textFile at :24scala> val rdd2 = rdd1.flatMap(_.split(" "))rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at flatMap at :26scala> val rdd3 = rdd2.map((_, 1))rdd3: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at :28scala> val rdd4 = rdd3.reduceByKey(_ + _)rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at :30

查看 RDD 的血缘关系

scala> rdd1.toDebugStringres0: String =(2) ./words.txt MapPartitionsRDD[1] at textFile at :24 []| ./words.txt HadoopRDD[0] at textFile at :24 []scala> rdd2.toDebugStringres1: String =(2) MapPartitionsRDD[2] at flatMap at :26 []| ./words.txt MapPartitionsRDD[1] at textFile at :24 []| ./words.txt HadoopRDD[0] at textFile at :24 []scala> rdd3.toDebugStringres2: String =(2) MapPartitionsRDD[3] at map at :28 []| MapPartitionsRDD[2] at flatMap at :26 []| ./words.txt MapPartitionsRDD[1] at textFile at :24 []| ./words.txt HadoopRDD[0] at textFile at :24 []scala> rdd4.toDebugStringres3: String =(2) ShuffledRDD[4] at reduceByKey at :30 []+-(2) MapPartitionsRDD[3] at map at :28 []| MapPartitionsRDD[2] at flatMap at :26 []| ./words.txt MapPartitionsRDD[1] at textFile at :24 [] | ./words.txt HadoopRDD[0] at textFile at :24 []

说明:

圆括号中的数字表示 RDD 的并行度. 也就是有几个分区.

查看 RDD 的依赖关系

scala> rdd1.dependenciesres4: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.OneToOneDependency@74d2525f)scala> rdd2.dependenciesres5: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.OneToOneDependency@401358e5)scala> rdd3.dependenciesres6: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.OneToOneDependency@4ed2eee1)scala> rdd4.dependenciesres7: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.ShuffleDependency@34341b2a)

想理解 RDDs 是如何工作的, 最重要的事情就是了解 transformations.

RDD 之间的关系可以从两个维度来理解: 一个是 RDD 是从哪些 RDD 转换而来, 也就是 RDD 的 parent RDD(s)是什么; 另一个就是 RDD 依赖于 parent RDD(s)的哪些 Partition(s). 这种关系就是 RDD 之间的依赖.

依赖 有 2 种策略:

1. 窄依赖(transformations with narrow dependencies)

2. 宽依赖(transformations with wide dependencies)

宽依赖对 Spark 去评估一个 transformations 有更加重要的影响, 比如对性能的影响.

窄依赖

如果 B RDD 是由 A RDD 计算得到的, 则 B RDD 就是 Child RDD, A RDD 就是 parent RDD.

如果依赖关系在设计的时候就可以确定, 而不需要考虑父 RDD 分区中的记录, 并且如果父 RDD 中的每个分区最多只有一个子分区, 这样的依赖就叫窄依赖

一句话总结: 父 RDD 的每个分区最多被一个 RDD 的分区使用

具体来说, 窄依赖的时候, 子 RDD 中的分区要么只依赖一个父 RDD 中的一个分区(比如map, filter操作), 要么在设计时候就能确定子 RDD 是父 RDD 的一个子集(比如: coalesce).

所以, 窄依赖的转换可以在任何的的一个分区上单独执行, 而不需要其他分区的任何信息.

宽依赖

如果 父 RDD 的分区被不止一个子 RDD 的分区依赖, 就是宽依赖.

宽依赖工作的时候, 不能随意在某些记录上运行, 而是需要使用特殊的方式(比如按照 key)来获取分区中的所有数据.

例如: 在排序(sort)的时候, 数据必须被分区, 同样范围的 key 必须在同一个分区内. 具有宽依赖的 transformations 包括: sort, reduceByKey, groupByKey, join, 和调用rePartition函数的任何操作.