Spark 键值对RDD数据分区

た入场券 2021-12-01 18:33 304阅读 0赞

# 3.键值对 RDD 数据分区 #

Spark 目前支持 Hash 分区和 Range 分区，用户也可以自定义分区，Hash 分区为当前

的默认分区，Spark 中分区器直接决定了 RDD 中分区的个数、RDD 中每条数据经过

Shuffle 过程属于哪个分区和 Reduce 的个数

![1604514-20190707165813180-1966062577.png][]

注意：

(1)只有 Key-Value 类型的 RDD 才有分区的，非 Key-Value 类型的 RDD 分区的值是 None

(2)每个 RDD 的分区 ID 范围：0~numPartitions-1，决定这个值是属于那个分区的。

## 3.1 获取 RDD 分区 ##

可以通过使用 RDD 的 partitioner 属性来获取 RDD 的分区方式。它会返回一个

scala.Option 对象， 通过 get 方法获取其中的值。相关源码如下：

def getPartition(key: Any): Int = key match {
      case null => 0
      case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
    }
      def nonNegativeMod(x: Int, mod: Int): Int = {
      val rawMod = x % mod
      rawMod + (if (rawMod < 0) mod else 0)
    }

（1）创建一个 pairRDD

scala> val pairs = sc.parallelize(List((1,1),(2,2),(3,3)))
    pairs: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[3] at parallelize at <console>:24

（2）查看 RDD 的分区器

scala> pairs.partitioner
    res1: Option[org.apache.spark.Partitioner] = None

（3）导入 HashPartitioner 类

scala> import org.apache.spark.HashPartitioner
    import org.apache.spark.HashPartitioner

（4）使用 HashPartitioner 对 RDD 进行重新分区

scala> val partitioned = pairs.partitionBy(new HashPartitioner(2))
    partitioned: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[4] at partitionBy at <console>:27

（5）查看重新分区后 RDD 的分区器

scala> partitioned.partitioner
    res2: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@2)

## 3.2 Hash 分区  ##

HashPartitioner 分区的原理：对于给定的 key，计算其 hashCode，并除以分区的个数

取余，如果余数小于 0，则用余数+分区的个数（否则加 0），最后返回的值就是这个 key 所

属的分区 ID。

使用 Hash 分区的实操

scala> nopar.partitioner
    res20: Option[org.apache.spark.Partitioner] = None
    
    scala> val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)
    nopar: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[10] at parallelize at <console>:24
    
    scala>nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+" : "+iter.mkString("|")) }).collect
    res0: Array[String] = Array("0 : ", 1 : (1,3), 2 : (1,2), 3 : (2,4), "4 : ", 5 : (2,3), 6 : (3,6), 7 : (3,8)) 
    
    scala> val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))
    hashpar: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[12] at partitionBy at <console>:26
    
    scala> hashpar.count
    res18: Long = 6
    
    scala> hashpar.partitioner
    res21: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@7)
    
    scala> hashpar.mapPartitions(iter => Iterator(iter.length)).collect()
    res19: Array[Int] = Array(0, 3, 1, 2, 0, 0, 0)

## 3.3 Ranger 分区  ##

HashPartitioner 分区弊端：可能导致每个分区中数据量的不均匀，极端情况下会导致

某些分区拥有 RDD 的全部数据。

RangePartitioner 作用：将一定范围内的数映射到某一个分区内，尽量保证每个分区

中数据量的均匀，而且分区与分区之间是有序的，一个分区中的元素肯定都是比另一个分

区内的元素小或者大，但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内

的数映射到某一个分区内。实现过程为：

第一步：先重整个 RDD 中抽取出样本数据，将样本数据排序，计算出每个分区的最

大 key 值，形成一个 Array\[KEY\]类型的数组变量 rangeBounds；

第二步：判断 key 在 rangeBounds 中所处的范围，给出该 key 值在下一个 RDD 中的

分区 id 下标；该分区器要求 RDD 中的 KEY 类型必须是可以排序的

## 3.4 自定义分区  ##

要实现自定义的分区器，你需要继承 org.apache.spark.Partitioner 类并实现下面三个

方法。

（1）numPartitions: Int:返回创建出来的分区数。

（2）getPartition(key: Any): Int:返回给定键的分区编号(0 到 numPartitions-1)。

（3）equals():Java 判断相等性的标准方法。这个方法的实现非常重要，Spark 需要用这个

方法来检查你的分区器对象是否和其他分区器实例相同，这样 Spark 才可以判断两个

RDD 的分区方式是否相同。

需求：将相同后缀的数据写入相同的文件，通过将相同后缀的数据分区到相同的分区并保

存输出来实现。

（1）创建一个 pairRDD

scala> val data = sc.parallelize(Array((1,1),(2,2),(3,3),(4,4),(5,5),(6,6)))
    data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[3] at parallelize at <console>:24

（2）定义一个自定义分区类

scala> :paste
    // Entering paste mode (ctrl-D to finish)
    class CustomerPartitioner(numParts:Int) extends org.apache.spark.Partitioner{
      //覆盖分区数
      override def numPartitions: Int = numParts
      //覆盖分区号获取函数
      override def getPartition(key: Any): Int = {
        val ckey: String = key.toString
        ckey.substring(ckey.length-1).toInt%numParts
      }
    }
    
    // Exiting paste mode, now interpreting.
    
    defined class CustomerPartitioner

（3）将 RDD 使用自定义的分区类进行重新分区

scala> val par = data.partitionBy(new CustomerPartitioner(2))
    par: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[2] at partitionBy at <console>:27

（4）查看重新分区后的数据分布

scala> par.mapPartitionsWithIndex((index,items)=>items.map((index,_))).collect
    res3: Array[(Int, (Int, Int))] = Array((0,(2,2)), (0,(4,4)), (0,(6,6)), (1,(1,1)), (1,(3,3)), (1,(5,5)))

使用自定义的 Partitioner 是很容易的:只要把它传给 partitionBy() 方法即可。

Spark 中有许多依赖于数据混洗的方法，比如 join() 和 groupByKey()，它们也可以

接收一个可选的 Partitioner 对象来控制输出数据的分区方式。

**笔记：**

package com.lxl
    
    import org.apache.spark.{Partitioner, SparkConf, SparkContext}
    
    class CustomerPartitioner(numPar: Int) extends Partitioner{
      override def numPartitions: Int = numPar
    
      override def getPartition(key: Any): Int = {
        key.toString.hashCode % numPar
      }
    }
    
    object App{
      def main(args: Array[String]): Unit = {
    
        //创建sparkConf对象
        val conf = new SparkConf().setAppName("partitioner").setMaster("local[*]")
    
        //创建sparkcontext对象
        val sc = new SparkContext(conf)
    
        val rdd = sc.makeRDD(Array(1,2,3,4,5))
    
        //查看原始分区存放方式
    //    rdd.mapPartitionsWithIndex((x, y) => Iterator(x + ":" + y.mkString("|"))).collect()
        rdd.mapPartitionsWithIndex((x, y) => Iterator(x + ":" + y.mkString("|"))).foreach(println)
    
        //查看自定义分区存放方式
    //    rdd.map((_,1)).partitionBy(new CustomerPartitioner(5)).mapPartitionsWithIndex((x, y) => Iterator(x + ":" + y.mkString("|"))).collect()
        rdd.map((_,1)).partitionBy(new CustomerPartitioner(5)).mapPartitionsWithIndex((x, y) => Iterator(x + ":" + y.mkString("|"))).foreach(println)
    
        sc.stop()
    
      }
    }

转载于:https://www.cnblogs.com/LXL616/p/11146982.html

[1604514-20190707165813180-1966062577.png]: /images/20211201/b8b75f61973944a085d2da0a00fc9da7.png