Spark – 一文搞懂 Partitioner

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

一.导言

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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 函数完成

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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}"))
      }
    })

Spark - 一文搞懂 Partitioner

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

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

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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}"))
      }
    })

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

三.RangePartitioner

1.源码剖析

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

除了 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 个分区。

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

B.依据Boundary获取分区

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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}"))
      }
    })

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

四.SelfPartitioner

1.源码剖析

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

numPartitions:获取总的分区数

getPartition: 获取 key 对应的分区 id

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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
      }
    }

Spark - 一文搞懂 Partitioner

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}"))
      }
    })

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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

1.函数剖析

Spark - 一文搞懂 Partitioner
Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

关于需求分区的 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
    }
  }

Spark - 一文搞懂 Partitioner

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}")
        })
      }
    }

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

这一届是带过最差的学生,咋还能考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
    }
  }

Spark - 一文搞懂 Partitioner

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

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

Spark - 一文搞懂 Partitioner

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

六.总结

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

Spark - 一文搞懂 Partitioner

Spark - 一文搞懂 Partitioner

发表评论

提供最优质的资源集合

立即查看 了解详情