RDD依赖关系及Stage划分源码

谁践踏了优雅 2022-12-31 06:25 116阅读 0赞

# RDD依赖关系 #

### 目录 ###

*  RDD依赖关系
 *   *  1. 查看血缘关系
     *  2. 查看依赖关系
     *  3. 窄依赖
     *  4. 宽依赖
     *  5. Stage任务划分
     *  6. Stage任务划分源码分析

## 1. 查看血缘关系 ##

> RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

1）代码实现

object Lineage01 { 
    
    def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            val fileRDD: RDD[String] = sc.textFile("input/1.txt")
            println(fileRDD.toDebugString)
            println("----------------------")
    
            val wordRDD: RDD[String] = fileRDD.flatMap(_.split(" "))
            println(wordRDD.toDebugString)
            println("----------------------")
    
            val mapRDD: RDD[(String, Int)] = wordRDD.map((_,1))
            println(mapRDD.toDebugString)
            println("----------------------")
    
            val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
            println(resultRDD.toDebugString)
    
            resultRDD.collect()
    
            //4.关闭连接
            sc.stop()
        }
    }

2）打印结果

(2) input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15 []
     |  input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15 []
    ----------------------
    (2) MapPartitionsRDD[2] at flatMap at Lineage01.scala:19 []
     |  input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15 []
     |  input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15 []
    ----------------------
    (2) MapPartitionsRDD[3] at map at Lineage01.scala:23 []
     |  MapPartitionsRDD[2] at flatMap at Lineage01.scala:19 []
     |  input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15 []
     |  input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15 []
    ----------------------
    (2) ShuffledRDD[4] at reduceByKey at Lineage01.scala:27 []
     +-(2) MapPartitionsRDD[3] at map at Lineage01.scala:23 []
        |  MapPartitionsRDD[2] at flatMap at Lineage01.scala:19 []
        |  input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15 []
        |  input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15 []

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

## 2. 查看依赖关系 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

1）代码实现

object Lineage02 { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            val fileRDD: RDD[String] = sc.textFile("input/1.txt")
            println(fileRDD.dependencies)
            println("----------------------")
    
            val wordRDD: RDD[String] = fileRDD.flatMap(_.split(" "))
            println(wordRDD.dependencies)
            println("----------------------")
    
            val mapRDD: RDD[(String, Int)] = wordRDD.map((_,1))
            println(mapRDD.dependencies)
            println("----------------------")
    
            val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
            println(resultRDD.dependencies)
    
            resultRDD.collect()
    
            // 查看localhost:4040页面，观察DAG图
    Thread.sleep(10000000)
    
            //4.关闭连接
            sc.stop()
        }
    }

2）打印结果

List(org.apache.spark.OneToOneDependency@f2ce6b)
    ----------------------
    List(org.apache.spark.OneToOneDependency@692fd26)
    ----------------------
    List(org.apache.spark.OneToOneDependency@627d8516)
    ----------------------
    List(org.apache.spark.ShuffleDependency@a518813)

3）全局搜索（ctrl+n）org.apache.spark.OneToOneDependency

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) { 
        override def getParents(partitionId: Int): List[Int] = List(partitionId)
    }

> **注意：要想理解RDDS是如何工作的，最重要的就是理解Transformations。  
> RDD之间的关系可以从两个维度来理解：一个是RDD是从哪些RDD转换而来，也就是 RDD的parent RDD(s)是什么; 另一个就是RDD依赖于parent RDD(s)的哪些Partition(s)，这种关系就是RDD之间的依赖。**

RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。

## 3. 窄依赖 ##

窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition使用，窄依赖我们形象的比喻为独生子女。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]

## 4. 宽依赖 ##

宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖，会引起Shuffle，总结：宽依赖我们形象的比喻为超生。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]

具有宽依赖的transformations包括：sort、reduceByKey、groupByKey、join和调用rePartition函数的任何操作。  
宽依赖对Spark去评估一个transformations有更加重要的影响，比如对性能的影响。

## 5. Stage任务划分 ##

**1）DAG有向无环图**  
DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。例如，DAG记录了RDD的转换过程和任务的阶段。  
![在这里插入图片描述][20201228205916606.png]

**2）任务运行的整体流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 5]

**3）RDD任务切分中间分为：Application、Job、Stage和Task**  
（1）Application：初始化一个SparkContext即生成一个Application；  
（2）Job：一个Action算子就会生成一个Job；  
（3）Stage：Stage等于宽依赖的个数加1；  
（4）Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。  
注意：Application->Job->Stage->Task每一层都是1对n的关系。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 6]

**4）代码实现**

object Stage01 { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2. Application：初始化一个SparkContext即生成一个Application；
            val sc: SparkContext = new SparkContext(conf)
    
            //3. 创建RDD
            val dataRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4,1,2),2)
    
            //3.1 聚合
            val resultRDD: RDD[(Int, Int)] = dataRDD.map((_,1)).reduceByKey(_+_)
    
            // Job：一个Action算子就会生成一个Job；
            //3.2 job1打印到控制台
            resultRDD.collect().foreach(println)
    
            //3.3 job2输出到磁盘
            resultRDD.saveAsTextFile("output")
    
            Thread.sleep(1000000)
    
            //4.关闭连接
            sc.stop()
        }
    }

**5）查看Job个数**  
查看http://localhost:4040/jobs/，发现Job有两个。  
![在这里插入图片描述][20201228210107145.png]

**6）查看Stage个数**  
查看Job0的Stage。由于只有1个Shuffle阶段，所以Stage个数为2。  
![在这里插入图片描述][20201228210117771.png]

查看Job1的Stage。由于只有1个Shuffle阶段，所以Stage个数为2。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 7]

**7）Task个数**  
查看Job0的Stage0的Task个数  
![在这里插入图片描述][20201228210133762.png]

查看Job0的Stage1的Task个数  
![在这里插入图片描述][20201228210137786.png]

查看Job1的Stage2的Task个数  
![在这里插入图片描述][2020122821014258.png]

查看Job1的Stage3的Task个数  
![在这里插入图片描述][2020122821014677.png]

注意：如果存在shuffle过程，系统会自动进行缓存，UI界面显示skipped的部分

## 6. Stage任务划分源码分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 8]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 9]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 10]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 11]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221120/e4a518358de14feaba39c760e1a1b647.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221120/88eac89ebd30492baaafac4aa323e432.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]: /images/20221120/2d6d6ca151f1430f919d429d61e5fcee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]: /images/20221120/0864902560424550975371a660170023.png
[20201228205916606.png]: /images/20221120/f021799b29e44ae1a4e648da27636501.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 4]: /images/20221120/26cf04525f7242d182cbb003df35c5bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 5]: /images/20221120/6f4b33f9512b4de09c14fab8c75caec0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 6]: /images/20221120/2010059b33e84a0ca43155f4f1b95358.png
[20201228210107145.png]: /images/20221120/395ab62066a6432e9176d93a45151901.png
[20201228210117771.png]: /images/20221120/1fa7dd77e4004fafb4ad095efff85f4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 7]: /images/20221120/99f22ac5cfe543089766057c8ff8bfd0.png
[20201228210133762.png]: /images/20221120/073f7e7e956343bf842338346c3c31e6.png
[20201228210137786.png]: /images/20221120/042f179cf78a414897dda8200dcee97e.png
[2020122821014258.png]: /images/20221120/75b7961ab1fc4df68048a58bc13e71b5.png
[2020122821014677.png]: /images/20221120/e2e6d4c6c0c540249fb7a7b13c12d373.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 8]: /images/20221120/7770edc2a25a440dab5e356d97cee313.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 9]: /images/20221120/55e9d5224682457088aab0f003505c5d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 10]: /images/20221120/65505634523742b5a4be605be04bf471.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 11]: /images/20221120/a431bb6a43f94df392d323d64b33f530.png