RDD数据分区器

简介: RDD数据分区器

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

Hash分区

HashPartitioner分区的原理:对于给定的key,计算其hashCode,并除以分区的个数取余,如果余数小于0,则用余数+分区的个数(否则加0),最后返回的值就是这个key所属的分区ID。
使用Hash分区的实操

scala> nopar.partitioner
res20: Option[org.apache.spark.Partitioner] = None

scala> val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)
nopar: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[10] at parallelize at <console>:24

scala>nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+" : "+iter.mkString("|")) }).collect
res0: Array[String] = Array("0 : ", 1 : (1,3), 2 : (1,2), 3 : (2,4), "4 : ", 5 : (2,3), 6 : (3,6), 7 : (3,8)) 
scala> val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))
hashpar: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[12] at partitionBy at <console>:26

scala> hashpar.count
res18: Long = 6

scala> hashpar.partitioner
res21: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@7)

scala> hashpar.mapPartitions(iter => Iterator(iter.length)).collect()
res19: Array[Int] = Array(0, 3, 1, 2, 0, 0, 0)

Ranger分区

HashPartitioner分区弊端:可能导致每个分区中数据量的不均匀,极端情况下会导致某些分区拥有RDD的全部数据。
RangePartitioner作用:将一定范围内的数映射到某一个分区内,尽量保证每个分区中数据量的均匀,而且分区与分区之间是有序的,一个分区中的元素肯定都是比另一个分区内的元素小或者大,但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。实现过程为:
第一步:先重整个RDD中抽取出样本数据,将样本数据排序,计算出每个分区的最大key值,形成一个Array[KEY]类型的数组变量rangeBounds;
第二步:判断key在rangeBounds中所处的范围,给出该key值在下一个RDD中的分区id下标;该分区器要求RDD中的KEY类型必须是可以排序的

自定义分区

要实现自定义的分区器,你需要继承 org.apache.spark.Partitioner 类并实现下面三个方法。
(1)numPartitions: Int:返回创建出来的分区数。
(2)getPartition(key: Any): Int:返回给定键的分区编号(0到numPartitions-1)。
(3)equals():Java 判断相等性的标准方法。这个方法的实现非常重要,Spark 需要用这个方法来检查你的分区器对象是否和其他分区器实例相同,这样 Spark 才可以判断两个 RDD 的分区方式是否相同。
需求:将相同后缀的数据写入相同的文件,通过将相同后缀的数据分区到相同的分区并保存输出来实现。
(1)创建一个pairRDD

scala> val data = sc.parallelize(Array((1,1),(2,2),(3,3),(4,4),(5,5),(6,6)))
data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[3] at parallelize at <console>:24

(2)定义一个自定义分区类

scala> :paste
// Entering paste mode (ctrl-D to finish)
class CustomerPartitioner(numParts:Int) extends org.apache.spark.Partitioner{

  //覆盖分区数
  override def numPartitions: Int = numParts

  //覆盖分区号获取函数
  override def getPartition(key: Any): Int = {
    val ckey: String = key.toString
    ckey.substring(ckey.length-1).toInt%numParts
  }
}

// Exiting paste mode, now interpreting.

defined class CustomerPartitioner

(3)将RDD使用自定义的分区类进行重新分区

scala> val par = data.partitionBy(new CustomerPartitioner(2))
par: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[2] at partitionBy at <console>:27

(4)查看重新分区后的数据分布

scala> par.mapPartitionsWithIndex((index,items)=>items.map((index,_))).collect
res3: Array[(Int, (Int, Int))] = Array((0,(2,2)), (0,(4,4)), (0,(6,6)), (1,(1,1)), (1,(3,3)), (1,(5,5)))

使用自定义的 Partitioner 是很容易的:只要把它传给 partitionBy() 方法即可。Spark 中有许多依赖于数据混洗的方法,比如 join() 和 groupByKey(),它们也可以接收一个可选的 Partitioner 对象来控制输出数据的分区方式。

目录
相关文章
|
算法 调度
操作系统中几种最常见的调度算法(适用于软件设计师考试与期末考试复习)
先进先出置换算法我们可以理解为排队准则(谁先来,谁就先运行)先来先到原则。该算法总是淘汰最先进入主页的页面,即选择在主存中驻留时间最久的页面淘汰。该算法简单,只要把一个进程调入主存的页面,然后按照先后链接一个队列,并设置一个指针即可。它是一个最直观、性能最差的算法,会有 Belady 现象(是指一个进程未分配它所要求的全部页面,有时就.....
1825 1
操作系统中几种最常见的调度算法(适用于软件设计师考试与期末考试复习)
|
Java Maven Spring
maven项目,动态读取外部配置文件
maven项目,动态读取外部配置文件
|
编解码 算法 数据处理
基于八叉树的空间划分及搜索操作
基于八叉树的空间划分及搜索操作
基于八叉树的空间划分及搜索操作
|
SQL 关系型数据库 MySQL
软件测试|使用PyMySQL访问MySQL数据库的详细指南
软件测试|使用PyMySQL访问MySQL数据库的详细指南
|
Java 数据安全/隐私保护
异常解决 javax.crypto.AEADBadTagException: Tag mismatch!
异常解决 javax.crypto.AEADBadTagException: Tag mismatch!
2674 0
异常解决 javax.crypto.AEADBadTagException: Tag mismatch!
|
12月前
|
机器学习/深度学习 算法 数据安全/隐私保护
基于yolov4深度学习网络的公共场所人流密度检测系统matlab仿真,带GUI界面
本项目使用 MATLAB 2022a 进行 YOLOv4 算法仿真,实现公共场所人流密度检测。通过卷积神经网络提取图像特征,将图像划分为多个网格进行目标检测和识别,最终计算人流密度。核心程序包括图像和视频读取、处理和显示功能。仿真结果展示了算法的有效性和准确性。
331 31
|
12月前
|
机器学习/深度学习 人工智能 算法
人工智能与机器人的结合:智能化世界的未来
人工智能与机器人的结合:智能化世界的未来
1473 32
|
11月前
|
存储 监控 算法
Flink 四大基石之 Checkpoint 使用详解
Flink 的 Checkpoint 机制通过定期插入 Barrier 将数据流切分并进行快照,确保故障时能从最近的 Checkpoint 恢复,保障数据一致性。Checkpoint 分为精确一次和至少一次两种语义,前者确保每个数据仅处理一次,后者允许重复处理但不会丢失数据。此外,Flink 提供多种重启策略,如固定延迟、失败率和无重启策略,以应对不同场景。SavePoint 是手动触发的 Checkpoint,用于作业升级和迁移。Checkpoint 执行流程包括 Barrier 注入、算子状态快照、Barrier 对齐和完成 Checkpoint。
2254 20
|
存储 缓存 算法
作者推荐 | 【深入浅出MySQL】「底层原理」探秘缓冲池的核心奥秘,揭示终极洞察
MySQL作为一个存储系统,有着一个关键的优化机制——缓冲池(buffer pool),它极大地提高了数据的访问效率,避免了频繁的磁盘IO操作。通过将常用的数据存储在内存中,MySQL可以快速响应查询请求,减少耗时的磁盘访问。这一优化机制在提升数据库性能方面起到了重要的作用。
703 7
作者推荐 | 【深入浅出MySQL】「底层原理」探秘缓冲池的核心奥秘,揭示终极洞察
|
算法 Python
Python 一步一步教你用pyglet制作汉诺塔游戏
Python 一步一步教你用pyglet制作汉诺塔游戏
417 0