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 []
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。