自定义分区器

比如说key是手机号,那么可以自定义分区器.

下面案例是根据key进行分区,偶数分到0号区域,奇数分到1号区域

  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}
  6. object Transformation_partitionBy {
  7.   def main(args: Array[String]): Unit = {
  9.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  11.     val sc: SparkContext = new SparkContext(conf)
  13.     val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)
  15.     rdd.mapPartitionsWithIndex {
  16.       (index, datas) => {
  17.         println(index + "------" + datas.mkString(","))
  18.         datas
  19.       }
  20.     }.collect()
  21.     /* 输出
  22.     0------(1,aaa)
  23.     2------(3,ccc)
  24.     1------(2,bbb)
  25.      */
  26.     println("----------------------------------------------")
  28.     val newRDD: RDD[(Int, String)] = rdd.partitionBy(new MyPartitioner(2))
  29.     newRDD.mapPartitionsWithIndex {
  30.       (index, datas) => {
  31.         println(index + "------" + datas.mkString(","))
  32.         datas
  33.       }
  34.     }.collect()
  35.     /*
  36.     输出
  37.     0------(2,bbb)
  38.     1------(1,aaa),(3,ccc)
  39.      */
  40.     // 关闭连接
  41.     sc.stop()
  43.   }
  44. }
  46. //自定义分区器
  47. class MyPartitioner(partitions: Int) extends Partitioner {
  49.   //获取分区个数
  50.   override def numPartitions: Int = partitions
  52.   /*指定分区规则
  53.     分区的时候,每个元素都会调用getPartition方法,返回int类型的值,这个int类型的值就是分区号, 从0开始的分区号
  54.   */
  56.   override def getPartition(key: Any): Int = {
  58.     val keyInt = key.toString.toInt
  59.     if (keyInt % 2 == 0) { //偶数
  60.       return 0
  61.     } else { // 奇数
  62.       return 1
  63.     }
  64.   }
  65. }