Spark 目前支持 Hash 分区和 Range 分区，和用户自定义分区。Hash 分区为当前的默认分区。分区器直接决定了 RDD 中分区的个数、RDD 中每条数据经过 Shuffle 后进入哪个分区，进而决定了 Reduce 的个数。
➢ 只有 Key-Value 类型的 RDD 才有分区器，非 Key-Value 类型的 RDD 分区的值是 None
➢ 每个 RDD 的分区 ID 范围：0 ~ (numPartitions - 1)，决定这个值是属于那个分区的。

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

  1. class HashPartitioner(partitions: Int) extends Partitioner {
  2.      require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")
  3.  def numPartitions: Int = partitions
  4.  def getPartition(key: Any): Int = key match {
  5.          case null => 0
  6.          case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
  7.      }
  8.  override def equals(other: Any): Boolean = other match {
  9.          case h: HashPartitioner =>
  10.          h.numPartitions == numPartitions
  11.          case _ =>
  12.              false
  13.  }
  14.  override def hashCode: Int = numPartitions
  15. }

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

  1. class RangePartitioner[K : Ordering : ClassTag, V](
  2.     partitions: Int,
  3.     rdd: RDD[_ <: Product2[K, V]],
  4.     private var ascending: Boolean = true,
  5.     val samplePointsPerPartitionHint: Int = 20)
  6.   extends Partitioner {
  8.   // A constructor declared in order to maintain backward compatibility for Java, when we add the
  9.   // 4th constructor parameter samplePointsPerPartitionHint. See SPARK-22160.
  10.   // This is added to make sure from a bytecode point of view, there is still a 3-arg ctor.
  11.   def this(partitions: Int, rdd: RDD[_ <: Product2[K, V]], ascending: Boolean) = {
  12.     this(partitions, rdd, ascending, samplePointsPerPartitionHint = 20)
  13.   }
  15.   // We allow partitions = 0, which happens when sorting an empty RDD under the default settings.
  16.   require(partitions >= 0, s"Number of partitions cannot be negative but found $partitions.")
  17.   require(samplePointsPerPartitionHint > 0,
  18.     s"Sample points per partition must be greater than 0 but found $samplePointsPerPartitionHint")
  20.   private var ordering = implicitly[Ordering[K]]
  22.   // An array of upper bounds for the first (partitions - 1) partitions
  23.   private var rangeBounds: Array[K] = {
  24.     if (partitions <= 1) {
  25.       Array.empty
  26.     } else {
  27.       // This is the sample size we need to have roughly balanced output partitions, capped at 1M.
  28.       // Cast to double to avoid overflowing ints or longs
  29.       val sampleSize = math.min(samplePointsPerPartitionHint.toDouble * partitions, 1e6)
  30.       // Assume the input partitions are roughly balanced and over-sample a little bit.
  31.       val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.length).toInt
  32.       val (numItems, sketched) = RangePartitioner.sketch(rdd.map(_._1), sampleSizePerPartition)
  33.       if (numItems == 0L) {
  34.         Array.empty
  35.       } else {
  36.         // If a partition contains much more than the average number of items, we re-sample from it
  37.         // to ensure that enough items are collected from that partition.
  38.         val fraction = math.min(sampleSize / math.max(numItems, 1L), 1.0)
  39.         val candidates = ArrayBuffer.empty[(K, Float)]
  40.         val imbalancedPartitions = mutable.Set.empty[Int]
  41.         sketched.foreach { case (idx, n, sample) =>
  42.           if (fraction * n > sampleSizePerPartition) {
  43.             imbalancedPartitions += idx
  44.           } else {
  45.             // The weight is 1 over the sampling probability.
  46.             val weight = (n.toDouble / sample.length).toFloat
  47.             for (key <- sample) {
  48.               candidates += ((key, weight))
  49.             }
  50.           }
  51.         }
  52.         if (imbalancedPartitions.nonEmpty) {
  53.           // Re-sample imbalanced partitions with the desired sampling probability.
  54.           val imbalanced = new PartitionPruningRDD(rdd.map(_._1), imbalancedPartitions.contains)
  55.           val seed = byteswap32(-rdd.id - 1)
  56.           val reSampled = imbalanced.sample(withReplacement = false, fraction, seed).collect()
  57.           val weight = (1.0 / fraction).toFloat
  58.           candidates ++= reSampled.map(x => (x, weight))
  59.         }
  60.         RangePartitioner.determineBounds(candidates, math.min(partitions, candidates.size))
  61.       }
  62.     }
  63.   }
  65.   def numPartitions: Int = rangeBounds.length + 1
  67.   private var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]
  69.   def getPartition(key: Any): Int = {
  70.     val k = key.asInstanceOf[K]
  71.     var partition = 0
  72.     if (rangeBounds.length <= 128) {
  73.       // If we have less than 128 partitions naive search
  74.       while (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) {
  75.         partition += 1
  76.       }
  77.     } else {
  78.       // Determine which binary search method to use only once.
  79.       partition = binarySearch(rangeBounds, k)
  80.       // binarySearch either returns the match location or -[insertion point]-1
  81.       if (partition < 0) {
  82.         partition = -partition-1
  83.       }
  84.       if (partition > rangeBounds.length) {
  85.         partition = rangeBounds.length
  86.       }
  87.     }
  88.     if (ascending) {
  89.       partition
  90.     } else {
  91.       rangeBounds.length - partition
  92.     }
  93.   }
  95.   override def equals(other: Any): Boolean = other match {
  96.     case r: RangePartitioner[_, _] =>
  97.       r.rangeBounds.sameElements(rangeBounds) && r.ascending == ascending
  98.     case _ =>
  99.       false
  100.   }
  102.   override def hashCode(): Int = {
  103.     val prime = 31
  104.     var result = 1
  105.     var i = 0
  106.     while (i < rangeBounds.length) {
  107.       result = prime * result + rangeBounds(i).hashCode
  108.       i += 1
  109.     }
  110.     result = prime * result + ascending.hashCode
  111.     result
  112.   }
  114.   @throws(classOf[IOException])
  115.   private def writeObject(out: ObjectOutputStream): Unit = Utils.tryOrIOException {
  116.     val sfactory = SparkEnv.get.serializer
  117.     sfactory match {
  118.       case js: JavaSerializer => out.defaultWriteObject()
  119.       case _ =>
  120.         out.writeBoolean(ascending)
  121.         out.writeObject(ordering)
  122.         out.writeObject(binarySearch)
  124.         val ser = sfactory.newInstance()
  125.         Utils.serializeViaNestedStream(out, ser) { stream =>
  126.           stream.writeObject(scala.reflect.classTag[Array[K]])
  127.           stream.writeObject(rangeBounds)
  128.         }
  129.     }
  130.   }
  132.   @throws(classOf[IOException])
  133.   private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
  134.     val sfactory = SparkEnv.get.serializer
  135.     sfactory match {
  136.       case js: JavaSerializer => in.defaultReadObject()
  137.       case _ =>
  138.         ascending = in.readBoolean()
  139.         ordering = in.readObject().asInstanceOf[Ordering[K]]
  140.         binarySearch = in.readObject().asInstanceOf[(Array[K], K) => Int]
  142.         val ser = sfactory.newInstance()
  143.         Utils.deserializeViaNestedStream(in, ser) { ds =>
  144.           implicit val classTag = ds.readObject[ClassTag[Array[K]]]()
  145.           rangeBounds = ds.readObject[Array[K]]()
  146.         }
  147.     }
  148.   }
  149. }

自定义分区器

  1. package part
  3. import org.apache.log4j.{Level, Logger}
  4. import org.apache.spark.{Partitioner, SparkConf, SparkContext}
  6. object Spark_Part {
  7.   def main(args: Array[String]): Unit = {
  8.       //屏蔽日志信息
  9.     Logger.getLogger("org").setLevel(Level.ERROR)
  10.     //创建sparkconf
  11.     val conf = new SparkConf().setMaster("local[2]").setAppName("wc")
  12.     //创建spark程序入口
  13.     val sc = new SparkContext(conf)
  14.     //创建集合对象
  15.     val list = List(("nba","************"),("cba","************"),
  16.                     ("wnba","************"),("nba","************"))
  17.     //将集合对象写进RDD里 并创建三个分区
  18.     val inputRDD = sc.makeRDD(list,3)
  19.     //将新的RDD使用partitionby方法自定义分区
  20.     val value = inputRDD.partitionBy(new Mypartitioner)
  21.     //保存到文件里
  22.     value.saveAsTextFile("output")
  24.     sc.stop()
  25.   }
  27.   /**
  28.    * 第一 : 自定义分区器
  29.    * 第二 : 重写方法
  30.    */
  32.   class Mypartitioner extends Partitioner{
  34.     //分区数量
  35.     override def numPartitions: Int = 3
  37.     //根据数据的key值 返回数据所在的分区索引 (从0开始)
  38.     override def getPartition(key: Any): Int = {
  39.       //方式一 : 用if做判断
  40. //      if(key == "nba"){
  41. //        0
  42. //      }else if (key == "cba"){
  43. //        1
  44. //      }else{
  45. //        2
  46. //      }
  48.       //方式二 : 用模式匹配
  49.       //如果是nba 放到0号分区，如果是cba 放到1号分区，如果是其他，放到2号分区
  50.       key match {
  51.         case "nba" => 0
  52.         case "cba" => 1
  53.         case _ => 2
  54.       }
  55.     }
  56.   }
  57. }