图解RDD血缘关系

• 需求

有三个RDD ,分别是 rddA,rddB,rddC.取数据1,2,3,4,5而且分红三个分区,对输入的数据的每个数据*2 ,只取大于 6 的数据.

• 代码

val rddA = sc.parallelize(List(1, 2, 3, 4, 5),3)
  //rddA: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0]
  val rddB = rddA.map(_*2)
  //rddB: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1]
  val rddC = rddB.filter(_>6)
  //rddC: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[2]
  rddC.collect()
  //res0: Array[Int] = Array(8, 10)

• 图解

• idea调试查看

• 经过 spark web ui 查看

• 补充

使用代码rddC.toDebugString打印依赖关系

res1: String =
(2) MapPartitionsRDD[2] at filter at <console>:25 []
 |  MapPartitionsRDD[1] at map at <console>:25 []
 |  ParallelCollectionRDD[0] at parallelize at <console>:24 []

窄依赖

• 说明:父RDD的每一个分区只被一个子RDD分区使用一次
• 窄依赖有分为两种：
• 1.一种是一对一的依赖，即OneToOneDependency
• 2.还有一个是范围的依赖RangeDependency，它仅仅被org.apache.spark.rdd.UnionRDD使用。UnionRDD是把多个RDD合成一个RDD，这些RDD是被拼接而成，每一个父RDD的Partition的相对顺序不会变，只不过每一个父RDD在UnionRDD中的Partition的起始位置不一样
• 常见算子

map, filter, union, join, mapPartitions, mapValues

• 图解
  

宽依赖

• 说明:父RDD的每一个分区都有可能被多个子RDD分区使用，子RDD分区一般对应父RDD全部分区
• 常见会对应Shuffle的操做.在会 job中产生一个stage
• groupByKey, join,partitionBy,reduce

• 常见算子

  • 图解

wordCountDemo演示

val path = "/user/spark/data/wc.txt"
    val lines = sc.textFile(path, 3)
    //查看每一个分区的数据
    //    lines.mapPartitionsWithIndex((n, partition) => {
    //      partition.map(x => (s"分区编号${n}", s"分区数据${x}"))
    //    }).foreach(println)
    
    val words = lines.flatMap(_.split(","))
    val wordPair = words.map(x => (x, 1))
    val result = wordPair.reduceByKey(_ + _)
    result.collect().foreach(println)

• 图解

若是以为文章不错的话记得关注下公号哦