Spark RDD深入浅析

Spark里的RDD是什么？在Spark如火如荼的今天，很多面试官都会问这个问题。想必答案大家都脱口而出--就是弹性分布式数据集嘛，但是它怎么就弹性了？它怎么分布式的？就需要去它的实现代码中一探究竟了。

RDD的实现本质

作为一个java类，其实RDD只有5个主要的属性

1. partition的集合。一般情况下，这个集合的大小决定了task的数量，因为每个partition会产生一个task。
2. 依赖的parent RDD的列表。RDD的依赖关系构成了RDD的lineage，是产生DAG的基础，也是容错处理的基础。
3. 给定一个partition作为参数，用来计算这个partition的函数。RDD的数据计算实际上在partition上并行进行的。这个函数可能是将parent的partition进行transform，也有可能是直接从外部读取数据
4. 一个可选的分区函数
5. 一个可选的preferred locations，用来达到计算局部性的目的。

所以虽然RDD叫做数据集，但是它并没有真正的存储数据，它只是存储了数据的来源和变换。 通过来源和变换，可以很容易的重新构建一个RDD，所以它是弹性的。 它的partition分布在不同的节点上，因此RDD也是分布式的。

RDD的变换和依赖

Spark中的transform，就是在现有RDD的基础上构建新的RDD的过程。新构建的RDD会将原有的RDD作为依赖，并且记录transform相应的变换。 transform不改变原有的RDD，所以在容错处理中，可以重复执行。

这样可以很容易的理解宽依赖和窄依赖。

宽依赖和窄依赖

窄依赖：parent RDD中的每个partition最多被子RDD中的一个partition使用 宽依赖：parent RDD中的每个partition被子RDD中的多个partition使用

在运行spark的时候，将参数spark.logLineage设置为true，就可以输出RDD的Lineage

object SparkApp {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("SparkApp")
      .config("spark.logLineage", true)
      .master("local")
      .getOrCreate()
    val rdd1 = spark.sparkContext.makeRDD(Seq(1, 2, 3, 4))
    val rdd2 = spark.sparkContext.makeRDD(Seq(10, 20, 30, 40))
    val result = rdd1.union(rdd2).map(_ * 2).reduce(_ + _)
    System.out.println(result)
    spark.stop()
  }
}

输出为

(2) MapPartitionsRDD[3] at map at SparkApp.scala:14 []
 |  UnionRDD[2] at union at SparkApp.scala:14 []
 |  ParallelCollectionRDD[0] at makeRDD at SparkApp.scala:12 []
 |  ParallelCollectionRDD[1] at makeRDD at SparkApp.scala:13 []