Spark – 一文搞懂 Partitioner-六虎

本文已参加「新人创造礼」活动，一起开启创造之路。

一.导言

spark 处理 RDD 时供给了 foreachPartition 和 mapPartitapplicationion 的办法对 partition 进行处理，一个 p二分查找时间复杂度artition 内或许包括一个文件或者多个文件的内容，Partitioner 能够依据 pair接口是什么RDD 的 key 完成自界说partappreciateition 的内容。

Partitioner 函数最根接口类型本的两个办法是 numP数组初始化artitions 和 getPartition(key: Any):

A.n数组umPartitions: 获二分查找算法取总的分区数

B.getParti接口类型tion:

依据 key 获取当时 partit二分查找平均查找长度公式ion 对应的分区数目，规模在 [0, numPartitions-1]，这儿的 partitionId 与 TaskContext.getPartitionId 的值共同，经过 hash(key) 得到 int 的 partitionNu数组去重方法m 变量，相同 partitonNum 的 keappley 对应的 paidRDD 将分到同一个 partition 内处理

常见接口类型的 Partition 分区类型有如下几种：

分区函数	分区办法
HashPartitioner	依据 hash(key) 分区
RangePartitioner	依据 Range 边界分区
Partitioner	依据自界说规矩分区

二.HashPartitioner

1.源码剖析

hashPartiti接口文档oner 依据 Object.has效率的英文hcode % partit接口是什么ionNum 进行分区，需求注意 partitio效率的英文nNum 的值是需求 >= 0 的，par数组初始化tiionNum 的获取经过 getPartition 函数内的 nonNegativeMod 函数完成

nonNegativeMod 在完成 hashCode% partappreciateitionNum接口卡 的基础上增加了非负性的要求，因为 partitionNum 是大于等于 0 的数目:

2.代码测试

    val testRdd = sc.parallelize((0 until 500).toArray.zipWithIndex)
    val partitionNum = 5
    testRdd.partitionBy(new HashPartitioner(partitionNum)).foreachPartition(partition => {
      if (partition.nonEmpty) {
        val info = partition.toArray.map(_._1)
        val taskId = TaskContext.getPartitionId()
        info.take(3).map(num => println(s"taskId: $taskId num: $num hashNum: ${num.hashCode()} %$partitionNum=${num.hashCode() % 5}"))
      }
    })

这儿将 0-499 共 500 个数字 zipWithIndex 生成 pairRdd 并经过 Hash二分查找Partitioner 生成 5 个 Partition，经过 TaskContext 获取 partitionId，为了日志逐个打印，这儿选用 local[1] 的配置 :

val conf = new SparkConf().setAppName("PartitionTest").setMaster("local[1]")

能够看到红框内效率是什么意思同一个 TaskId 对应的 partiti数组排序on 内的 key 都具有相同的 mod 值，所以被分到同一分区。

3.repartition

正常运用的 repartition 函数选用 HashPartitioner 函数作为默许分区函数，下面尝试一下：

    println("=============================repartition=============================")
    testRdd.repartition(5).foreachPartition(partition => {
      if (partition.nonEmpty) {
        val info = partition.toArray.map(_._1)
        val taskId = TaskContext.getPartitionId()
        info.take(3).map(num => println(s"taskId: $taskId num: $num hashNum: ${num.hashCode()} %$partitionNum=${num.hashCode() % 5}"))
      }
    })

与上面不同的是 taskId 有差异，可是相同 mod 的 key 仍然会分到同一分区：

三.RangePartitioner

1.源码剖析

RangePartitioner 依据规模将元素大致均匀的分配至不同分区 partition，规模appear经过传入 RDD 的内容采样来确定。

除了 partitions 的参数外，RangePartitioner 还需求将待分区的 rdd 传入供随机采样生成 rangeBounds 运二分查找算法用，相比于 HashPartition 直接接口 h接口类型ashCodes接口crc错误计数 % partitionNum 的操作，RangePartitioner 分区共分两步：

A.获取分区 Boundary

需求采样的分区样本巨细上线为 1m，转换为 double 防止精度溢出，第一个 else 逻辑内考虑假如一个分区内包括的项目数远远超过平均数，则从中从头效率计算公式采样，以确保该分区能够收集到足够的采样数目，最下面的 if 函数运用所需的采样概率对不平衡分效率符号区从头采样，终究得到appear分区的边界，这儿能够抽空单独效率是什么意思拎出来研究一下。举个例子大致理解下，假如所有 partition 内的 key 的规模是 0-500，随机生成5效率公式个分区，则生成 101-203-299-405 这样的区间，每一个数字代表其分区的上界，例如分区0的上界为 101，分区1的上界为 203，依次类推，终究生成 5 个分区。

B.依据Boundary获取分区

假如分区数组长度不大于 128，则进行简略的暴力循环查找，假如超过 128，则进行二分查找，一起供给依据 ascending 参数决议 partitionId 的次序或逆序。这与之前 Spark-Scala 数据特征分桶时选用的优化战略共同，有兴趣能够看看：Scala – 数值型特征分桶。

2.代码测试

    val testRdd = sc.parallelize((0 until 500).toArray.zipWithIndex)
    testRdd.partitionBy(new RangePartitioner(5, testRdd)).foreachPartition(partition => {
      if (partition.nonEmpty) {
        val info = partition.toArray.map(_._1)
        val taskId = TaskContext.getPartitionId()
        info.take(3).map(num => println(s"taskId: $taskId num: $num hashNum: ${num.hashCode()} length: ${info.length}"))
      }
    })

仍然运用 500 个纯数字 RDD 进行 range 分区的测试，为了验证大致均分的思维，这儿最终不再打印 mod 成果，转而打印每个 partition 内元素的数量，能够看到数组指针这次每组数量不像之前 HashPartitioner 得到的相同均匀，而是介于 500/5=100 的上下，可是总数为 500。

四.SelfPartitioner

1.源码剖析

自界说 Partitioner 首要完成下述两个功能，上面也提到了，再简略补充下:

numPartitions:获取总的分区数

getPartition: 获取 key 对应的分区 id

2.代码完成

A.SelfPartitioner

依据上面 RangePartitioner 分区二分查找不均匀的状况，咱们选用 SelfParitioner 自界说分区的办法完成均匀分区，这儿偷接口测试懒直接生成了对应的上界 boundary，实际场景中 boundary 应该依据 partitionNum 的数量动态生成，getPartition 函数内引入了 break 机制，不熟悉的同学能够移步：Scala – 优雅的break，随后就是基础的暴力循环，假如找到上界则回来上界对应的 index 作为分区 id。

    import scala.util.control.Breaks._
    class SelfPartition(partitionNum: Int) extends Partitioner {
      val boundary: Array[(Int, Int)] = Array(100, 200, 300, 400, 500).zipWithIndex
      override def numPartitions: Int = partitionNum
      override def getPartition(key: Any): Int = {
        val keyNum = key.toString.toInt
        var partitionNum = 0
        breakable(
          boundary.foreach(bound => {
            if (keyNum < bound._1) {
              partitionNum = bound._2
              break()
            }
          })
        )
        partitionNum
      }
    }

B.运转成果

    val testRdd = sc.parallelize((0 until 500).toArray.zipWithIndex)
    testRdd.partitionBy(new SelfPartition(5)).foreachPartition(partition => {
      if (partition.nonEmpty) {
        val info = partition.toArray.map(_._1)
        val taskId = TaskContext.getPartitionId()
        info.take(3).map(num => println(s"taskId: $taskId num: $num hashNum: ${num.hashCode()} length: ${info.length}"))
      }
    })

经过 SelfPartitioner 分区后能效率集够看到 0-499 共 500 个元素被均匀分配到 5 个 Part二分查找算法it二分查找代码ion 内，除了效率最简略的 boundary 办法分区外，也能够自界说 hash 办法，key 的类型默许为 Any，假如 key 不是 scala 的根本数据类型，则运用 key.asInstanceOf[T] 转换即可。

五.repar接口测试用例设计titionAndSortWithinParti数组c语言tions

1.函数剖析

除了正常的分区需求外，spark 还供给repartitionAndSortWithinParti接口文档tions 函数，该函数依据数组词给定的分区器 Partitiappointmentoner 进行分区区分得到新的 RDD，并依据每个键进行排序，使得 RDD 中的数据坚持必定次序，该办法比 repartition + so数组去重rting 愈加高效，因为它把排序机制放入了 shuffle接口类型的过程中。

源码中该办法位二分查找时间复杂度于 OrderedRddFunctions 类内，只支撑传入分区函数 Partitioner，ordering 排序规矩需求运用 implict 传入隐函数的办法界说：

关于需求分区的 key: Any，需求界说隐接口和抽象类的区别函数保证其完成 Ordering 接口才能完成分区后排序，不然只接口测试用例设计能分区没有排序。

2.代码完成

A.分区排序依据数组排序

在分区函数的基础上，增加了 Ordering 隐函数，这儿 Partitioner 函数仍然担任依据 key 得到分区 Id，和上面不同的时，分区的一起对 rdd 进行 shuffle，其中 order 的规矩由隐函数给出，这儿经过比较二者的分数进行排二分查找时间复杂度序，假如想要逆序只需求添加负号即可-(接口自动化x.score – y.score)。

  // 学生类
  case class Student(name: String, score: Int)
  // 隐函数-Ordering
  implicit object StudentOrdering extends Ordering[Student] {
    override def compare(x: Student, y: Student): Int = {
      x.score - y.score
    }
  }
  class SelfSortPartition(partitionNum: Int) extends Partitioner {
    val boundary: Array[(Int, Int)] = Array(100, 200, 300, 400, 500).zipWithIndex
    override def numPartitions: Int = partitionNum
    override def getPartition(key: Any): Int = {
      val stuKey = key.asInstanceOf[Student]
      val keyNum = stuKey.name.toInt
      var partitionNum = 0
      breakable(
        boundary.foreach(bound => {
          if (keyNum < bound._1) {
            partitionNum = bound._2
            break()
          }
        })
      )
      partitionNum
    }
  }

B.主函数

这儿运用 0-499 的 String 类型作为学生的编号，Score 则采纳 ma接口自动化th.random x 100 进行模仿，分区运用 Student 的 name id，所以每个元素的分区不变，变的是每个元appreciate素的次序。

    println("=============================SortPartition=============================")
    val studentRdd = sc.parallelize((0 until 500).toArray.map(num => (Student(num.toString, (math.random * 100).toInt), true)))
    studentRdd.take(5).foreach(println(_))
    studentRdd.repartitionAndSortWithinPartitions(new SelfSortPartition(5)).foreachPartition(partition => {
      if (partition.nonEmpty) {
        val taskId = TaskContext.getPartitionId()
        println("===========================")
        partition.toArray.take(5).map(stu => {
          println(s"TaskId: ${taskId} Name: ${stu._1.name} Score: ${stu._1.score}")
        })
      }
    }

因为appstore运用 x.score – y.score 次序计数，所以按分数从小到大排序:

这一届是带过最差的学生，咋还能考0分。

C.其他写法

除了 StudentOrdering 的写法，也能够选用直接 Object Student 的写法，这儿 A <: Student 表明任何 A 的子类都支撑该隐式调用，关于 <: 相关知识能够参考：Scala Generic 泛型类详解 – T

  object Student {
    implicit def orderingByGradeStudentScore[A <: Student]: Ordering[A] = {
      Ordering.by(stu => stu.score)
    }
  }
  implicit object StudentOrdering extends Ordering[Student] {
    override def compare(x: Student, y: Student): Int = {
      x.score - y.score
    }
  }

假如想要支撑多重排序，能够在元祖内增加多个字段，会优先比较 name 再比较 score，以此类推，同理假如想要逆序，例如依据分数逆序摆放，则改为 (stu.name，-1 * stu.score)

  object Student {
    implicit def orderingByGradeStudentScore[A <: Student]: Ordering[A] = {
      Ordering.by(stu => (stu.name, stu.score))
    }
  }

假如对应的分区 key 没有完成 implict 的比较隐函数，则函数会直接报灰，无法编译：

六.总结

Partitioner 的一般用法大致就这些，除了三种 HashPartitioner 函数外，Spark 也能够经过repartitionAndSortWit效率hin二分查找算法举例说明Paappetitertitions 完成分区 + 排序的需求，整体来说，Parti效率公式tioner 支二分查找代码撑用户自界说分区规矩去规划任务的 task 需求处理什么样接口crc错误计数的 partition 数据，关于精细化数组处理approve和区域化定制的需求非常方便，除此之外，一些需求次序处理的数据或者次序存储的数据，经过 SortWithinPart数组和链表的区别itions 的办法也能够进步功率，非常奈斯。最终继续感叹命名的接口卡抽象性，partition – 分片、隔墙，现在看到屏风就像看到了 RDD。

Spark – 一文搞懂 Partitioner

一.导言

二.HashPartitioner

1.源码剖析

2.代码测试

3.repartition

三.RangePartitioner

1.源码剖析

2.代码测试

四.SelfPartitioner

1.源码剖析

2.代码完成

五.repar接口测试用例设计titionAndSortWithinParti数组c语言tions

1.函数剖析

2.代码完成

六.总结

相关文章

一文读懂 Android FFmpeg 视频解码过程与实战分析

[别被脱裤系列]2 还没深入数据库就浅出了

Flutter是如何渲染Widget的

谁说 Java 不能多继承

作者信息