Spark-尚硅谷5-2 数据结构：RDD 行动算子

分手后的思念是犯贱 2022-10-22 07:52 183阅读 0赞

# 5.1.4.5 RDD 行动算子 #

// TODO - 行动算子
        // 所谓的行动算子，其实就是触发作业(Job)执行的方法
        // 底层代码调用的是环境对象的runJob方法
        // 底层代码中会创建ActiveJob，并提交执行。

## 1) reduce ##

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

## 2) collect ##

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

**实现wordcount的方法**

package com.atguigu.bigdata.spark.core.wc
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ SparkConf, SparkContext}
    
    import scala.collection.mutable
    
    object Spark03_WordCount { 
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparConf)
    
            wordcount91011(sc)
    
            sc.stop()
    
        }
    
        // groupBy
        def wordcount1(sc : SparkContext): Unit = { 
    
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val group: RDD[(String, Iterable[String])] = words.groupBy(word=>word)
            val wordCount: RDD[(String, Int)] = group.mapValues(iter=>iter.size)
        }
    
        // groupByKey
        def wordcount2(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordOne = words.map((_,1))
            val group: RDD[(String, Iterable[Int])] = wordOne.groupByKey()
            val wordCount: RDD[(String, Int)] = group.mapValues(iter=>iter.size)
        }
    
        // reduceByKey
        def wordcount3(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordOne = words.map((_,1))
            val wordCount: RDD[(String, Int)] = wordOne.reduceByKey(_+_)
        }
    
        // aggregateByKey
        def wordcount4(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordOne = words.map((_,1))
            val wordCount: RDD[(String, Int)] = wordOne.aggregateByKey(0)(_+_, _+_)
        }
    
        // foldByKey
        def wordcount5(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordOne = words.map((_,1))
            val wordCount: RDD[(String, Int)] = wordOne.foldByKey(0)(_+_)
        }
    
        // combineByKey
        def wordcount6(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordOne = words.map((_,1))
            val wordCount: RDD[(String, Int)] = wordOne.combineByKey(
                v=>v,
                (x:Int, y) => x + y,
                (x:Int, y:Int) => x + y
            )
        }
    
        // countByKey 行动算子
        def wordcount7(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordOne = words.map((_,1))
            val wordCount: collection.Map[String, Long] = wordOne.countByKey()
        }
    
        // countByValue 行动算子
        def wordcount8(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
            val wordCount: collection.Map[String, Long] = words.countByValue()
        }
    
        // reduce, aggregate, fold 行动算子
        def wordcount91011(sc : SparkContext): Unit = { 
            val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
            val words = rdd.flatMap(_.split(" "))
    
            // 【（word, count）,(word, count)】
            // word => Map[(word,1)]
            val mapWord = words.map(
                word => { 
                    mutable.Map[String, Long]((word,1))
                }
            )
    
           val wordCount = mapWord.reduce(
                (map1, map2) => { 
                    map2.foreach{ 
                        case (word, count) => { 
                            val newCount = map1.getOrElse(word, 0L) + count
                            map1.update(word, newCount)
                        }
                    }
                    map1
                }
            )
    
            println(wordCount)
        }
    
    }

## 10）save 相关算子 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]

// 保存成 Text 文件
    rdd.saveAsTextFile("output")
    // 序列化成对象保存到文件
    rdd.saveAsObjectFile("output1")
    // 保存成 Sequencefile 文件，要求数据的格式必须为K-V类型
    rdd.map((_,1)).saveAsSequenceFile("output2")

package com.atguigu.bigdata.spark.core.rdd.operator.action
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark05_RDD_Operator_Action { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
            val sc = new SparkContext(sparkConf)
    
            //val rdd = sc.makeRDD(List(1,1,1,4),2)
            val rdd = sc.makeRDD(List(
                ("a", 1),("a", 2),("a", 3)
            ))
    
            // TODO - 行动算子
            rdd.saveAsTextFile("output")
            rdd.saveAsObjectFile("output1")
            // saveAsSequenceFile方法要求数据的格式必须为K-V类型
            rdd.saveAsSequenceFile("output2")
            sc.stop()
    
        }
    }

## 11) foreach ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
    // 收集后打印
    rdd.map(num=>num).collect().foreach(println)
    println("****************")
    // 分布式打印
    rdd.foreach(println)

**算子 ： Operator（操作）  
RDD的方法和Scala集合对象的方法不一样  
Scala集合对象的方法都是在同一个节点的内存中完成的。  
RDD的方法可以将计算逻辑发送到Executor端（分布式节点）执行， 为了和Scala集合对象区分不同的处理效果，所以将RDD的方法称之为算子。  
RDD的方法外部的操作都是在Driver端执行的，而方法内部的逻辑代码是在Executor端执行。**

import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark06_RDD_Operator_Action { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
            val sc = new SparkContext(sparkConf)
    
            val rdd = sc.makeRDD(List(1,2,3,4))
    
            // Driver端打印输出的：foreach 其实是Driver端内存集合的循环遍历方法，打印出来是有序的
            rdd.collect().foreach(println)
            println("******************")
            // Executor端打印输出的：foreach 其实是Executor端内存数据打印(分布式打印)，打印出来是无序的
            rdd.foreach(println)
    
            sc.stop()
    
        }
    }

1
    2
    3
    4
    ******************
    2
    1
    3
    4

// Driver端打印输出的：foreach 其实是Driver端内存集合的循环遍历方法，打印出来是有序的
            rdd.collect().foreach(println)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 4]

rdd.foreach(println) // Executor端打印输出的：foreach 其实是Executor端内存数据打印(分布式打印)，打印出来是无序的
            rdd.foreach(println)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 5]

# RDD 序列化 #

Drive->Exectour 在网络中传输对象要序列化

package com.atguigu.bigdata.spark.core.rdd.operator.action
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark07_RDD_Operator_Action { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
            val sc = new SparkContext(sparkConf)
    
            val rdd = sc.makeRDD(List[Int](1,2,3,4))
    
            val user = new User()
            // RDD算子中传递的函数是会包含闭包操作，那么就会进行检测功能
            // 闭包检测
            rdd.foreach(
                num => { 
                    println("age = " + (user.age + num))
                }
            )
            sc.stop()
        }
        class User extends Serializable { 
        // 样例类在编译时，会自动混入序列化特质（实现可序列化接口）
        //case class User() { 
        //class User { 
            var age : Int = 30
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 6]

## 1) 闭包检查 ##

RDD算子中传递的函数是会包含闭包操作，那么就会进行检测功能  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 7]

## 2) 序列化方法和属性 ##

![在这里插入图片描述][20210406193927522.png]

object serializable02_function { 
     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)
     //3.创建一个 RDD
     val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", 
    "hive", "atguigu"))
     //3.1 创建一个 Search 对象
     val search = new Search("hello")
     //3.2 函数传递，打印：ERROR Task not serializable
     search.getMatch1(rdd).collect().foreach(println)
     //3.3 属性传递，打印：ERROR Task not serializable
     search.getMatch2(rdd).collect().foreach(println)
     //4.关闭连接
     sc.stop()
     } }
    class Search(query:String) extends Serializable { 
     def isMatch(s: String): Boolean = { 
     s.contains(query)
     }
     // 函数序列化案例
     def getMatch1 (rdd: RDD[String]): RDD[String] = { 
     //rdd.filter(this.isMatch)
     rdd.filter(isMatch)
     }
      // 属性序列化案例
     def getMatch2(rdd: RDD[String]): RDD[String] = { 
     //rdd.filter(x => x.contains(this.query))
     rdd.filter(x => x.contains(query))
     //val q = query
     //rdd.filter(x => x.contains(q))
     } }

package com.atguigu.bigdata.spark.core.rdd.serial
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark01_RDD_Serial { 
    
        def main(args: Array[String]): Unit = { 
            val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparConf)
    
            val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))
    
            val search = new Search("h")
    
            //search.getMatch1(rdd).collect().foreach(println)
            search.getMatch2(rdd).collect().foreach(println)
    
            sc.stop()
        }
        // 查询对象
        // 类的构造参数其实是类的属性, 构造参数需要进行闭包检测，其实就等同于类进行闭包检测
        class Search(query:String) extends Serializable { 
    
            def isMatch(s: String): Boolean = { 
                s.contains(this.query)
            }
    
            // 函数序列化案例
            def getMatch1 (rdd: RDD[String]): RDD[String] = { 
                rdd.filter(isMatch)
            }
    
            // 属性序列化案例
            def getMatch2(rdd: RDD[String]): RDD[String] = { 
                val s = query
                rdd.filter(x => x.contains(s))
            }
        }
    }

## 3) Kryo 序列化框架 ##

参考地址: https://github.com/EsotericSoftware/kryo  
![在这里插入图片描述][20210406195508550.png]  
Spark 内部已经使用Kryo所以我们自己就不需要使用了。

object serializable_Kryo { 
     def main(args: Array[String]): Unit = { 
     val conf: SparkConf = new SparkConf()
     .setAppName("SerDemo")
     .setMaster("local[*]")
     // 替换默认的序列化机制
     .set("spark.serializer", 
    "org.apache.spark.serializer.KryoSerializer")
     // 注册需要使用 kryo 序列化的自定义类
     .registerKryoClasses(Array(classOf[Searcher]))
     val sc = new SparkContext(conf)
     val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", 
    "atguigu", "hahah"), 2)
     val searcher = new Searcher("hello")
     val result: RDD[String] = searcher.getMatchedRDD1(rdd)
     result.collect.foreach(println)
     } }
    case class Searcher(val query: String) { 
     def isMatch(s: String) = { 
     s.contains(query)
     }
     def getMatchedRDD1(rdd: RDD[String]) = { 
     rdd.filter(isMatch) 
     }
     def getMatchedRDD2(rdd: RDD[String]) = { 
     val q = query
     rdd.filter(_.contains(q))
     } }

# 5.1.4.7 RDD 依赖关系 #

## 1) RDD 血缘关系 ##

![在这里插入图片描述][20210406202333521.png]

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()

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 8]

## 2) RDD 依赖关系 ##

这里所谓的依赖关系，其实就是两个相邻 RDD 之间的关系

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()

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 9]

![在这里插入图片描述][2021040620370685.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 10]

## 3) RDD 窄依赖 （一对以） ##

![在这里插入图片描述][20210406202431473.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 11]

![在这里插入图片描述][2021040620565596.png]

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd)

## 4) RDD 宽依赖 （shuffle 一对多 ） ##

![在这里插入图片描述][20210406205243902.png]  
一对多  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 12]  
![在这里插入图片描述][20210406205628340.png]

窄依赖和宽依赖的区别  
窄依赖：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 13]

宽依赖  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 14]

class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
     @transient private val _rdd: RDD[_ <: Product2[K, V]],
     val partitioner: Partitioner,
     val serializer: Serializer = SparkEnv.get.serializer,
     val keyOrdering: Option[Ordering[K]] = None,
     val aggregator: Option[Aggregator[K, V, C]] = None,
     val mapSideCombine: Boolean = false)
     extends Dependency[Product2[K, V]]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 15]

## 5) RDD 阶段划分 ##

依赖关系oneToOne不需要划分阶段  
依赖关系一对多（shuffle）需要划分阶段  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 16]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 17]

## 6) RDD 阶段划分源码 ##

package com.atguigu.bigdata.spark.core.rdd.dep
    
    import org.apache.spark.{ SparkConf, SparkContext}
    import org.apache.spark.rdd.RDD
    
    object Spark01_RDD_Dep { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparConf)
    
            val lines: RDD[String] = sc.textFile("datas/word.txt")
            println(lines.toDebugString)
            println("*************************")
            val words: RDD[String] = lines.flatMap(_.split(" "))
            println(words.toDebugString)
            println("*************************")
            val wordToOne = words.map(word=>(word,1))
            println(wordToOne.toDebugString)
            println("*************************")
            val wordToSum: RDD[(String, Int)] = wordToOne.reduceByKey(_+_)
            println(wordToSum.toDebugString)
            println("*************************")
            val array: Array[(String, Int)] = wordToSum.collect()
            array.foreach(println)
    
            sc.stop()
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 18]  
源码：

// New stage creation may throw an exception if, for example, jobs are run on 
    a
     // HadoopRDD whose underlying HDFS files have been deleted.
     finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
    } catch { 
     case e: Exception =>
     logWarning("Creating new stage failed due to exception - job: " + jobId, e)
     listener.jobFailed(e)
     return
    }
    ……
    private def createResultStage(
     rdd: RDD[_],
     func: (TaskContext, Iterator[_]) => _,
     partitions: Array[Int],
     jobId: Int,
     callSite: CallSite): ResultStage = { 
    val parents = getOrCreateParentStages(rdd, jobId)
    val id = nextStageId.getAndIncrement()
    val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
    stageIdToStage(id) = stage
    updateJobIdStageIdMaps(jobId, stage)
    stage
    
    ……
    private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] 
    = { 
    getShuffleDependencies(rdd).map {  shuffleDep =>
     getOrCreateShuffleMapStage(shuffleDep, firstJobId)
    }.toList
    }
    ……
    private[scheduler] def getShuffleDependencies(
     rdd: RDD[_]): HashSet[ShuffleDependency[_, _, _]] = { 
    val parents = new HashSet[ShuffleDependency[_, _, _]]
    val visited = new HashSet[RDD[_]]
    val waitingForVisit = new Stack[RDD[_]]
    waitingForVisit.push(rdd)
    while (waitingForVisit.nonEmpty) { 
     val toVisit = waitingForVisit.pop()
     if (!visited(toVisit)) { 
     visited += toVisit
     toVisit.dependencies.foreach { 
     case shuffleDep: ShuffleDependency[_, _, _] =>
     parents += shuffleDep
     case dependency =>
     waitingForVisit.push(dependency.rdd)
     }
     } }
    parents
    }
    }

## 7) RDD 任务划分 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 19]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 20]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 21]

## 8) RDD 任务划分源码 ##

val tasks: Seq[Task[_]] = try { 
     stage match { 
     case stage: ShuffleMapStage =>
     partitionsToCompute.map {  id =>
     val locs = taskIdToLocations(id)
     val part = stage.rdd.partitions(id)
     new ShuffleMapTask(stage.id, stage.latestInfo.attemptId,
     taskBinary, part, locs, stage.latestInfo.taskMetrics, properties, 
    Option(jobId),
     Option(sc.applicationId), sc.applicationAttemptId)
     }
     case stage: ResultStage =>
     partitionsToCompute.map {  id =>
     val p: Int = stage.partitions(id)
     val part = stage.rdd.partitions(p)
     val locs = taskIdToLocations(id)
     new ResultTask(stage.id, stage.latestInfo.attemptId,
     taskBinary, part, locs, id, properties, stage.latestInfo.taskMetrics,
     Option(jobId), Option(sc.applicationId), sc.applicationAttemptId)
     }
     }
    ……
    val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()
    ……
    override def findMissingPartitions(): Seq[Int] = { 
    mapOutputTrackerMaster
     .findMissingPartitions(shuffleDep.shuffleId)
     .getOrElse(0 until numPartitions) }

# 5.1.4.8 RDD 持久化 #

1.  cache(内存中) :

*  将数据临时存储在内存中进行数据重用
 *  会在血缘关系中添加新的依赖。一旦，出现问题，可以重头读取数据

1.  persist(临时存储在磁盘) :

*  将数据临时存储在磁盘文件中进行数据重用  
    \- 涉及到磁盘IO，性能较低，但是数据安全
    
     *  如果作业执行完毕，临时保存的数据文件就会丢失
     *  会在血缘关系中添加新的依赖。一旦，出现问题，可以重头读取数据

1.  checkpoint(长久存储在磁盘) :

*  将数据长久地保存在磁盘文件中进行数据重用
 *  涉及到磁盘IO，性能较低，但是数据安全
 *  为了保证数据安全，所以一般情况下，会独立执行作业(多次从头到尾执行任务)效率就会低，为了能够提高效率，一般情况下，是需要和cache联合使用（ mapRDD.cache() mapRDD.checkpoint()），就可以执行一次
 *  执行过程中，会切断血缘关系。重新建立新的血缘关系，及checkpoint等同于改变数据源，因为checkpoint是永久保存很安全，可以直接当做数据源。
 *  checkpoint 一般都是保存在分布式存储系统中例如HDFS
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 22]  
    **想要共用需要把RDD中的数据存储一下**  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 23]

package com.atguigu.bigdata.spark.core.rdd.persist
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark06_RDD_Persist { 
    
        def main(args: Array[String]): Unit = { 
    
    1. cache(内存中) : 
    - 将数据临时存储在内存中进行数据重用
    - 会在血缘关系中添加新的依赖。一旦，出现问题，可以重头读取数据
    3.  persist(临时存储在磁盘) : 将数据临时存储在磁盘文件中进行数据重用
            // 涉及到磁盘IO，性能较低，但是数据安全
            // 如果作业执行完毕，临时保存的数据文件就会丢失
    4. checkpoint(长久存储在磁盘) : 将数据长久地保存在磁盘文件中进行数据重用
            // 涉及到磁盘IO，性能较低，但是数据安全
            // 为了保证数据安全，所以一般情况下，会独立执行作业
            // 为了能够提高效率，一般情况下，是需要和cache联合使用
            // 执行过程中，会切断血缘关系。重新建立新的血缘关系
            // checkpoint等同于改变数据源
            // checkpoint 一般都是保存在分布式存储系统中例如HDFS
    
            val sparConf = new SparkConf().setMaster("local").setAppName("Persist")
            val sc = new SparkContext(sparConf)
            sc.setCheckpointDir("cp")
    
            val list = List("Hello Scala", "Hello Spark")
    
            val rdd = sc.makeRDD(list)
    
            val flatRDD = rdd.flatMap(_.split(" "))
    
            val mapRDD = flatRDD.map(word=>{ 
                (word,1)
            })
            // 将数据临时存储在内存中进行数据重用
           // 会在血缘关系中添加新的依赖。一旦，出现问题，可以重头读取数据
            //mapRDD.cache()
            // persist : 将数据临时存储在磁盘文件中进行数据重用
            // 涉及到磁盘IO，性能较低，但是数据安全
            // 如果作业执行完毕，临时保存的数据文件就会丢失
            //mapRDD.persist();
            mapRDD.checkpoint()
            println(mapRDD.toDebugString)
            val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_)
            reduceRDD.collect().foreach(println)
            println("**************************************")
            println(mapRDD.toDebugString)
            sc.stop()
        }
    }

## 1) RDD Cache 缓存 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 24]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 25]

## 2) RDD CheckPoint 检查点 ##

![在这里插入图片描述][20210408200652828.png]

// 设置检查点路径
    sc.setCheckpointDir("./checkpoint1")
    // 创建一个 RDD，读取指定位置文件:hello atguigu atguigu
    val lineRdd: RDD[String] = sc.textFile("input/1.txt")
    // 业务逻辑
    val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    val wordToOneRdd: RDD[(String, Long)] = wordRdd.map { 
     word => { 
     (word, System.currentTimeMillis())
     } }
    // 增加缓存,避免再重新跑一个 job 做 checkpoint
    wordToOneRdd.cache()
    // 数据检查点：针对 wordToOneRdd 做检查点计算
    wordToOneRdd.checkpoint()
    // 触发执行逻辑
    wordToOneRdd.collect().foreach(println)

## 3) 缓存和检查点区别 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 26]

# 5.1.4.9 RDD 分区器 #

![在这里插入图片描述][2021040820301281.png]

## 1) Hash 分区 ##

Hash 分区：对于给定的 key，计算其 hashCode,并除以分区个数取余

class HashPartitioner(partitions: Int) extends Partitioner { 
     require(partitions >= 0, s"Number of partitions ($partitions) cannot be 
    negative.")
     def numPartitions: Int = partitions
     def getPartition(key: Any): Int = key match { 
     case null => 0
     case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
     }
     override def equals(other: Any): Boolean = other match { 
     case h: HashPartitioner =>
     h.numPartitions == numPartitions
     case _ =>
     false
     }
     override def hashCode: Int = numPartitions
    }

## 2) Range 分区： ##

Range 分区：将一定范围内的数据映射到一个分区中，尽量保证每个分区数据均匀，而  
且分区间有序

class RangePartitioner[K : Ordering : ClassTag, V](
     partitions: Int,
     rdd: RDD[_ <: Product2[K, V]],
     private var ascending: Boolean = true)
     extends Partitioner { 
     // We allow partitions = 0, which happens when sorting an empty RDD under the 
    default settings.
     require(partitions >= 0, s"Number of partitions cannot be negative but found 
    $partitions.")
     private var ordering = implicitly[Ordering[K]]
     // An array of upper bounds for the first (partitions - 1) partitions
     private var rangeBounds: Array[K] = { 
     ...
     }
     def numPartitions: Int = rangeBounds.length + 1
     private var binarySearch: ((Array[K], K) => Int) = 
    CollectionsUtils.makeBinarySearch[K]
     def getPartition(key: Any): Int = { 
     val k = key.asInstanceOf[K]
     var partition = 0
     if (rangeBounds.length <= 128) { 
     // If we have less than 128 partitions naive search
     while (partition < rangeBounds.length && ordering.gt(k, 
    rangeBounds(partition))) { 
     partition += 1
     }
     } else { 
     // Determine which binary search method to use only once.
     partition = binarySearch(rangeBounds, k)
     // binarySearch either returns the match location or -[insertion point]-1
     if (partition < 0) { 
     partition = -partition-1
     }
     if (partition > rangeBounds.length) { 
     partition = rangeBounds.length
     }
     }
     if (ascending) { 
     partition
     } else { 
     rangeBounds.length - partition
     }
     }
     override def equals(other: Any): Boolean = other match { 
     ...
     }
     override def hashCode(): Int = { 
     ...
     }
     @throws(classOf[IOException])
     private def writeObject(out: ObjectOutputStream): Unit = 
    Utils.tryOrIOException { 
     ...
     }
     @throws(classOf[IOException])
     private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException 
    { 
     ...
     } }

## 3）自定分区器 ##

继承Partitioner 实现自定分区器

package com.atguigu.bigdata.spark.core.rdd.part
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ HashPartitioner, Partitioner, SparkConf, SparkContext}
    
    object Spark01_RDD_Part { 
    
        def main(args: Array[String]): Unit = { 
            val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.makeRDD(List(
                ("nba", "xxxxxxxxx"),
                ("cba", "xxxxxxxxx"),
                ("wnba", "xxxxxxxxx"),
                ("nba", "xxxxxxxxx"),
            ),3)
            val partRDD: RDD[(String, String)] = rdd.partitionBy( new MyPartitioner )
    
            partRDD.saveAsTextFile("output")
    
            sc.stop()
        }
    
        /** * 自定义分区器 * 1. 继承Partitioner * 2. 重写方法 */
        class MyPartitioner extends Partitioner{ 
            // 分区数量
            override def numPartitions: Int = 3
    
            // 根据数据的key值返回数据所在的分区索引（从0开始）
            override def getPartition(key: Any): Int = { 
                key match { 
                    case "nba" => 0
                    case "wnba" => 1
                    case _ => 2
                }
            }
        }
    }

# 5.1.4.10 RDD 文件读取与保存 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 27]  
读取

package com.atguigu.bigdata.spark.core.rdd.io
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark01_RDD_IO_Load { 
    
        def main(args: Array[String]): Unit = { 
            val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.textFile("output1")
            println(rdd.collect().mkString(","))
    
            val rdd1 = sc.objectFile[(String, Int)]("output2")
            println(rdd1.collect().mkString(","))
    
            val rdd2 = sc.sequenceFile[String, Int]("output3")
            println(rdd2.collect().mkString(","))
    
            sc.stop()
        }
    }

保存

package com.atguigu.bigdata.spark.core.rdd.io
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark01_RDD_IO_Save { 
    
        def main(args: Array[String]): Unit = { 
            val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparConf)
    
            val rdd = sc.makeRDD(
                List(
                    ("a", 1),
                    ("b", 2),
                    ("c", 3)
                )
            )
    
            rdd.saveAsTextFile("output1")
            rdd.saveAsObjectFile("output2")
            rdd.saveAsSequenceFile("output3")
            sc.stop()
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20221022/90dade9665204ce9b0635133be9f5308.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]: /images/20221022/c8e7091cbc2c4743bd53d7676e017ee5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]: /images/20221022/01c19d521f7f49d0a952873ff218ad45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]: /images/20221022/c44a0d138e754abaa29e5cbade2115fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 4]: /images/20221022/69a5a70754664829905e97e3636697be.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 5]: /images/20221022/1ab60cc917b04a7a89c2835b75c1ee92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 6]: /images/20221022/f4077e36fc25497ab22a13928a0564e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 7]: /images/20221022/8b38afa1e75b42e297f112331517b0dc.png
[20210406193927522.png]: /images/20221022/3d0a9efe2a8e48bf89e09d940809a198.png
[20210406195508550.png]: /images/20221022/46294bdde07c446ebd01849960a9bc1f.png
[20210406202333521.png]: /images/20221022/6275a96c5c204be1bbc696ca2777e47b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 8]: /images/20221022/cec39687447b4313872b9b71648aea5c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 9]: /images/20221022/c17432f526b34aaf93395fb622cb2320.png
[2021040620370685.png]: /images/20221022/b1de64990c174746915ec771f0cf90d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 10]: /images/20221022/5bf16ced3c9b4eeba80030b1939cab01.png
[20210406202431473.png]: /images/20221022/aa672222f19148ddb26eb08d766f5ac2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 11]: /images/20221022/def6ac3a55ef4f32b5344f3a18cbd1f9.png
[2021040620565596.png]: /images/20221022/68c886ebad3d4c77bbe3e255ca4dd138.png
[20210406205243902.png]: /images/20221022/230662b3951c42228c4b157e1ee5543e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 12]: /images/20221022/87118df3fef5450784fcd165148366b7.png
[20210406205628340.png]: /images/20221022/dffda0447b18478baf03cd8eecde9f9f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 13]: /images/20221022/cf8376ea445a4117992fa788a7e836a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 14]: /images/20221022/8de28c5adfda40ddbfe84d59ff7ed157.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 15]: /images/20221022/1b4c7d0948e448b79d2cd197bd3fb2f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 16]: /images/20221022/4689ae1e10d64ee7a5a13ea27d32194c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 17]: /images/20221022/e3ac9f7fb7a943d59fa9ec1f0d71e386.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 18]: /images/20221022/ae95faa8dea34f1585190f4ccff77eac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 19]: /images/20221022/01e330c76e384161aec6941c47afc762.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 20]: /images/20221022/16ab7a0db298435eaecf123007b2ec81.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 21]: /images/20221022/b088adb118a0493aa963bd77dd41c9c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 22]: /images/20221022/807c39876de5448bb75baa6e583521a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 23]: /images/20221022/5fc55730f44f4cf2bf96f8abef1ba1b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 24]: /images/20221022/fd3c55c8e16c45adbee94e630ef8ed40.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 25]: /images/20221022/71fb4192dbbb47d5a6a0f95ec6f1d340.png
[20210408200652828.png]: /images/20221022/49d66fbe174a4177aafc0c13fd7a9ce4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 26]: /images/20221022/3a1b20ecb4f44c76be30a2b1824675d9.png
[2021040820301281.png]: /images/20221022/5f19c434a40a4acab6eee6d2fdec75f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 27]: /images/20221022/8a81aa04f19740abaf4a0cb70c54f4b2.png