Spark RDD宽窄依赖

谁借莪１个温暖的怀抱￠ 2024-04-17 22:06 61阅读 0赞

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

### 1.窄依赖 ###

窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用

总结：窄依赖我们形象的比喻为**独生子女**

*  父RDD和子RDD partition之间的关系是**一对一**的。
 *  父RDD和子RDD partition之间的关系是**多对一**的。
 *  **不会有shuffle的产生**

### 2.宽依赖(shuffle依赖) ###

宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition

**宽依赖一定会引起shuffle**

**shuffle:将相同数据从不同节点拉到一起**

总结：宽依赖我们形象的比喻为**超生**

一个父RDD的结果去往多个子RDD的不同partition

父RDD与子RDD partition之间的关系是**一对多**。

**宽依赖算子**:groupByKey、reduceByKey,join

**会发生shuffle的算子:**

1.  **byKey类的算子**:如reduceByKey、groupByKey、 sortByKey、 aggregateByKey、combineByKey
2.  **repartition类的算子**:如repartition (少量分区变成多个分区会发生shuffle)、coalesce (需要指定是否发生shuffle)、repartitionAndSortWithinPartitions、partitionBy
3.  **join类的算子**:如join (先groupByKey后再join就不会发生shuffle)、cogroup, 这里需要注意，对于上述操作，能不使用shuffle操作，就尽量不用，尽量使用不发生shuffle的操作。另外,如果使用了shuffle操作，那么必须要进行shuffle的调优，甚至是解决遇到的数据倾斜问题。
    
    Join 不一定意味着宽依赖和 shuffle，但宽依赖一定伴随着 shuffle

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

### 3.宽窄依赖图理解 ###

宽依赖往往对应着shuffle操作，需要在运行过程中将同一个父RDD的分区传入到不同的子RDD分区中，中间可能涉及多个节点之间的数据传输；而窄依赖的每个父RDD的分区只会传入到一个子RDD分区中，通常可以在一个节点内完成转换。

当RDD分区丢失时（某个节点故障），spark会对数据进行重算

*  **对于窄依赖**，由于父RDD的一个分区只对应一个子RDD分区，这样只需要重算和子RDD分区对应的父RDD分区即可，所以这个**重算对数据的利用率是100%的**。
 *  **对于宽依赖**，重算的父RDD分区对应多个子RDD分区，这样实际上父RDD 中只有一部分的数据是被用于恢复这个丢失的子RDD分区的，另一部分对应子RDD的其它未丢失分区，这就造成了多余的计算；更一般的，宽依赖中子RDD分区通常来自多个父RDD分区，**极端情况下，所有的父RDD分区都要进行重新计算**。
 *  如下图所示，b1分区丢失，则需要重新计算a1,a2和a3，这就产生了冗余计算(a1,a2,a3中对应b2的数据)。
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2ODM4MzE1_size_16_color_FFFFFF_t_70 1]

**窄依赖的函数有**：map, filter, union, join(父RDD是hash-partitioned ), mapPartitions, mapValues

**宽依赖的函数有**：groupByKey, join(父RDD不是hash-partitioned ), partitionBy

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

### 4.区分宽依赖和窄依赖的好处 ###

1.  首先，narrow dependencies可以支持在同一个cluster node上，以pipeline形式执行多条命令，例如在执行了map后，紧接着执行filter。
    
    相反，shuffle / wide dependencies 需要所有的父分区都是可用的，可能还需要调用类似MapReduce之类的操作进行跨节点传递。
2.  其次，则是从失败恢复的角度考虑。 narrow dependencies的失败恢复更有效，因为它只需要重新计算丢失的parent partition即可，而且可以并行地在不同节点进行重计算。
    
    相反，shuffle / wide dependencies 牵涉RDD各级的多个parent partition。

### 5.宽依赖和窄依赖的实现 ###

所有的依赖都要实现trait Dependency\[T\]：

abstract class Dependency[T] extends Serializable { 
        def rdd: RDD[T]
    }

其中rdd就是依赖的parent RDD。

#### 5.1 对于窄依赖的实现（两种） ####

abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] { 
        //返回子RDD的partitionId依赖的所有的parent RDD的Partition(s)
        def getParents(partitionId: Int): Seq[Int]
        override def rdd: RDD[T] = _rdd
    }

窄依赖是有两种具体实现，分别如下：

一种是一对一的依赖，即 **OneToOneDependency**：

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

通过getParents的实现不难看出，RDD仅仅依赖于parent RDD相同ID的Partition。

还有一个是范围的依赖，即 **RangeDependency**，它仅仅被org.apache.spark.rdd.UnionRDD使用。UnionRDD是把多个RDD合成一个RDD，这些RDD是被拼接而成，即每个parent RDD的Partition的相对顺序不会变，只不过每个parent RDD在UnionRDD中的Partition的起始位置不同。因此它的getPartents如下：

override def getParents(partitionId: Int) = { 
        if(partitionId >= outStart && partitionId < outStart + length) { 
           List(partitionId - outStart + inStart)
        } else { 
           Nil
        }
    }

其中，inStart是parent RDD中Partition的起始位置，outStart是在UnionRDD中的起始位置，length就是parent RDD中Partition的数量。

#### 5.2 宽依赖的实现（只有一种） ####

宽依赖的实现只有一种：**ShuffleDependency**。子RDD依赖于parent RDD的所有Partition，因此需要Shuffle过程：

class ShuffleDependency[K, V, C](
        @transient _rdd: RDD[_ <: Product2[K, V]],
        val partitioner: Partitioner,
        val serializer: Option[Serializer] = None,
        val keyOrdering: Option[Ordering[K]] = None,
        val aggregator: Option[Aggregator[K, V, C]] = None,
        val mapSideCombine: Boolean = false)
    extends Dependency[Product2[K, V]] { 
     
    override def rdd = _rdd.asInstanceOf[RDD[Product2[K, V]]]
    //获取新的shuffleId
    val shuffleId: Int = _rdd.context.newShuffleId()
    //向ShuffleManager注册Shuffle的信息
    val shuffleHandle: ShuffleHandle =
    _rdd.context.env.shuffleManager.registerShuffle(
        shuffleId, _rdd.partitions.size, this)
     
        _rdd.sparkContext.cleaner.foreach(_.registerShuffleForCleanup(this))
    }

**注意**：宽依赖支持两种Shuffle Manager。

即 org.apache.spark.shuffle.hash.HashShuffleManager（基于Hash的Shuffle机制）

和 org.apache.spark.shuffle.sort.SortShuffleManager（基于排序的Shuffle机制）。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2ODM4MzE1_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/4d87712226bf40adb67437740462e51f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2ODM4MzE1_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/b4b60c7d71e5437e8ca9b1455354c73b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2ODM4MzE1_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/7044c620f51242e686f10ea01996dffb.png