0. 相关文章链接

Spark文章汇总

1. RDD 血缘关系

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

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * @ date: 2023/7/4
 * @ author: yangshibiao
 * @ desc: 项目描述
 */
object ModelTest {

    def main(args: Array[String]): Unit = {

        val conf: SparkConf = new SparkConf()
            .setAppName("ModelTest")
            .setMaster("local[*]")

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

    }
}

2. RDD依赖关系

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


import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * @ date: 2023/7/4
 * @ author: yangshibiao
 * @ desc: 项目描述
 */
object ModelTest {

    def main(args: Array[String]): Unit = {

        val conf: SparkConf = new SparkConf()
            .setAppName("ModelTest")
            .setMaster("local[*]")

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

    }
}

3. RDD窄依赖

窄依赖表示每一个父(上游)RDD 的 Partition 最多被子（下游）RDD 的一个 Partition 使用，窄依赖我们形象的比喻为独生子女。

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

6. RDD宽依赖

宽依赖表示同一个父（上游）RDD 的 Partition 被多个子（下游）RDD 的 Partition 依赖，会引起 Shuffle，总结：宽依赖我们形象的比喻为多生。

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

7. RDD阶段划分

DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。例如，DAG 记录了 RDD 的转换过程和任务的阶段。

8. RDD阶段划分源码

try { 
  // 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 
}