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 []

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券