Spark的RDD磁區器

RDD 磁區器

基本介紹

Spark 目前支持Hash 磁區、Range 磁區和用戶自定義磁區，Hash 磁區為當前的默認磁區，磁區器直接決定了RDD 中磁區的個數、RDD 中每條資料經過Shuffle 后進入哪個磁區，進而決定了Reduce 的個數，

（1）只有Key-Value 型別的RDD 才有磁區器，非 Key-Value 型別的RDD 磁區的值是 None，
（2）每個RDD 的磁區 ID 范圍：0 ~ (numPartitions - 1)，決定這個值是屬于那個磁區的，

1. 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磁區

說明

將一定范圍內的資料映射到一個磁區中，盡量保證每個磁區資料均勻，而且磁區間有序，

原始碼

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. 用戶自定義磁區

說明

用戶可以根據自己的需要，自定義磁區個數，

案例實操

package com.atguigu.bigdata.spark.core.rdd.part

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

object Spark01_RDD_Part {

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark01_RDD_Part")
    val sc = new SparkContext(sparkConf)

    val rdd = sc.makeRDD(List(
      ("nba","xxxxxxx"),
      ("cba","xxxxxxx"),
      ("wnba","xxxxxxx"),
      ("nba","xxxxxxx")
    ),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 "cba" => 2
        case _ => 2
      }
//      if (key == "nba"){
//        0
//      }else if ( key == "wnba"){
//       1
//      }else if (key == "cba"){
//        2
//      }else {
//        2
//      }
    }
  }

}