【大数据学习 | Spark-Core】Spark的分区器(HashPartitioner和RangePartitioner)

server/2024/11/25 12:15:10/

之前学过的kv类型上面的算子

groupby groupByKey reduceBykey sortBy sortByKey join[cogroup left inner right] shuffle的

mapValues keys values flatMapValues 普通算子,管道形式的算子

shuffle的过程是因为数据产生了打乱重分,分组、排序、join等算子需要将数据重新排版。

shuffle的过程是上游的数据处理完毕写出到自己的磁盘上,然后下游的数据从磁盘上面拉取。

重新排版打乱重分是需要存在规则的。

中间数据的流向规则叫做分区器 partitioner,这个分区器一般是存在于shuffle类算子中的,我们可以这么说,shuffle类算子一定会带有分区器,分区器也可以单独存在,人为定义分发规则。

groupBy groupBykey reduceBykey 自带的分区器HashPartitioner。

sortby sortBykey rangePartitioner

hashPartitioner

规则 按照key的hashCode %下游分区 = 分区编号

处理key-value类型数据,如果key为0,就分配去0号分区。否则调用nonNegativeMod函数。

保证取余的结果为正向结果。

hash取余的方式,不管数据分发到下游的什么分区中,最终结果都是相同的数据放入到一起。

演示结果:

scala> val arr = Array(1,2,3,4,5,6,7,8,9)
arr: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)scala> sc.makeRDD(arr,3)
res78: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[73] at makeRDD at <console>:27scala> res78.mapPartitionsWithIndex((index,it)=> it.map((index,_)))
res79: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[74] at mapPartitionsWithIndex at <console>:26scala> res79.collect
res80: Array[(Int, Int)] = Array((0,1), (0,2), (0,3), (1,4), (1,5), (1,6), (2,7), (2,8), (2,9))scala> res78.map(t=>(t,t))
res81: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[75] at map at <console>:26scala> res78.partitioner
res82: Option[org.apache.spark.Partitioner] = Nonescala> res81.reduceByKey(_+_)
res84: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[76] at reduceByKey at <console>:26scala> res84.partitioner
res85: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@3)scala> res84.mapPartitionsWithIndex((index,it)=> it.map((index,_)))
res88: org.apache.spark.rdd.RDD[(Int, (Int, Int))] = MapPartitionsRDD[77] at mapPartitionsWithIndex at <console>:26scala> res88.collect
res89: Array[(Int, (Int, Int))] = Array((0,(6,6)), (0,(3,3)), (0,(9,9)), (1,(4,4)), (1,(1,1)), (1,(7,7)), (2,(8,8)), (2,(5,5)), (2,(2,2)))

演示的逻辑,就是按照key.hashcode进行分区,int类型的hashcode值是自己的本身。

并且hash分区器的规则致使我们可以任意的修改下游的分区数量。

scala> res81.reduceByKey(_+_,100)
res91: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[78] at reduceByKey at <console>:26scala> res91.partitions.size
res92: Int = 100scala> res81.reduceByKey(_+_,2)
res93: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[79] at reduceByKey at <console>:26scala> res93.partitions.size
res94: Int = 2

rangePartitioner

hashPartitioner规则非常简单,直接规定来一个数据按照hashcode规则的分配,规则比较简答,但是会出现数据倾斜。

range分区规则中存在两个方法。

rangeBounds界限,在使用这个分区器之前先做一个界限划定。

首先使用水塘抽样算法,在未知的数据集中抽取能够代表整个数据集的样本,根据样本进行规则设定。

然后在使用getPartitions。

首先存在水塘抽样,规定数据的流向以后再执行整体逻辑,会先触发计算。

sortBykey是转换类的算子,不会触发计算。

但是我们发现它触发计算了,因为首先在计算之前先进行水塘抽样,能够规定下游的数据规则,然后再进行数据的计算。

scala> arr
res101: Array[Int] = Array(1, 9, 2, 8, 3, 7, 4, 6, 5)scala> arr.map(t=> (t,t))
res102: Array[(Int, Int)] = Array((1,1), (9,9), (2,2), (8,8), (3,3), (7,7), (4,4), (6,6), (5,5))scala> sc.makeRDD(res102)
res104: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[94] at makeRDD at <console>:27scala> res104.sortByKey()
res105: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[97] at sortByKey at <console>:26scala> res105.partitioner
res106: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.RangePartitioner@fe1f9dea)scala> res104.sortByKey(true,3)
res107: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[100] at sortByKey at <console>:26scala> res107.mapPartitionsWithIndex((index,it)=> it.map((index,_)))
res109: org.apache.spark.rdd.RDD[(Int, (Int, Int))] = MapPartitionsRDD[101] at mapPartitionsWithIndex at <console>:26scala> res109.collect
res110: Array[(Int, (Int, Int))] = Array((0,(1,1)), (0,(2,2)), (0,(3,3)), (1,(4,4)), (1,(5,5)), (1,(6,6)), (2,(7,7)), (2,(8,8)), (2,(9,9)))

range分区器,它是先做抽样然后指定下游分区的数据界限。

它可以修改分区数量,但是分区数量不能大于元素个数,必须要保证每个分区中都有元素。

scala> res104.sortByKey(true,3)
res111: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[104] at sortByKey at <console>:26scala> res104.sortByKey(true,300)
res112: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[107] at sortByKey at <console>:26scala> res111.partitions.size
res114: Int = 3scala> res112.part

自定义分区器

工作的过程中我们会遇见数据分类的情况,想要根据自己的需求定义分区的规则,让符合规则的数据发送到不同的分区中,这个时候我们就需要自定义分区器了。

定义分区器,让数据发送到不同的分区,从而不同的task任务输出的文件结果内容也不同

# 自己创建数据data/a.txt
hello tom hello jack
hello tom hello jack
hello tom hello jack
hello tom hello jack
hello tom hello jack
# 需求就是将数据按照规则进行分发到不同的分区中
# 存储的时候一个文件存储hello其他的文件存储tom jack

分区器的定义需要实现分区器的接口

class MyPartitioner extends Partitioner{override def numPartitions: Int = ???
// 设定下游存在几个分区override def getPartition(key: Any): Int = ???
// 按照key设定分区的位置
}

整体代码:

package com.hainiu.sparkimport org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}object Test1 {def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("parse")conf.setMaster("local[*]")val sc = new SparkContext(conf)val rdd = sc.textFile("data/a.txt").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)val rdd1 = rdd.partitionBy(new MyPartitioner)val fs = FileSystem.get(new Configuration())val out = "data/res"if(fs.exists(new Path(out)))fs.delete(new Path(out),true)rdd1.saveAsTextFile(out)}
}
class MyPartitioner extends Partitioner{override def numPartitions: Int = 2override def getPartition(key: Any): Int = {if(key.toString.equals("hello"))0else1}
}


http://www.ppmy.cn/server/144793.html

相关文章

【汇编】有关AI人工智能

随着 AI 技术的不断发展&#xff0c;AI大模型正在重塑软件开发流程&#xff0c;从代码自动生成到智能测试&#xff0c;未来&#xff0c;AI 大模型将会对软件开发者、企业&#xff0c;以及整个产业链都产生深远的影响。欢迎与我们一起&#xff0c;从 AI 大模型的定义、应用场景、…

java抽奖系统(一)

1. 项⽬介绍 1.1 背景 随着数字营销的兴起&#xff0c;企业越来越重视通过在线活动来吸引和留住客⼾。抽奖活动作为⼀种有效的营 销⼿段&#xff0c;能够显著提升⽤⼾参与度和品牌曝光率。于是我们就开发了以抽奖活动作为背景的Spring Boot项⽬&#xff0c;通过这个项⽬提供⼀…

mysql约束和高级sql

约束 MySQL中的约束用于定义表中数据的规则&#xff0c;以确保数据的准确性和可靠性。以下是MySQL中常用的一些约束类型及其概述&#xff1a; PRIMARY KEY&#xff08;主键&#xff09;&#xff1a;唯一标识表中每条记录的字段或字段组合&#xff0c;一个表中只能有一个主键。…

第2.3 AI文本—prompt入门

AIGC&#xff08;生成式人工智能&#xff09;对我们教师工作帮助是非常大的。我们教师要充分认识AIGC技术在辅助教学工作、行政工作、个人晋升成长、日常工作的重要性&#xff0c;并能熟练掌握国内主流的AIGC工具&#xff0c;最终达成教学工作提质增效目的。 辅助教学工作&…

PCA算法所体现的核心数学思维

一、PCA算法的基本思想 PCA算法的核心思想是通过线性变换&#xff0c;将数据从原始的高维空间投影到低维空间&#xff0c;同时尽可能保留数据的主要变异性。这种变换是通过找到一组新的坐标轴&#xff08;即主成分&#xff09;来实现的&#xff0c;这些坐标轴是原始数据空间的…

深入解析信号量:定义与环形队列生产消费模型剖析

一、定义 信号量&#xff0c;从本质上来说&#xff0c;它是一种特殊的计数器。其核心作用在于对临界资源中资源数量进行精确的描述与把控。以电影院为例&#xff0c;观众们会竞相选择不同的座位&#xff0c;而这些座位作为共享资源&#xff0c;被拆分成了多个独立的部分。只…

以太坊Beam Chain;后量子聚合签名;

目录 以太坊Beam Chain 愿景 技术层构建 区块生产: 质押: 密码学: 后量子聚合签名 后量子聚合签名简介 使用哈希函数实现后量子聚合签名 简单举例说明 优势 以太坊Beam Chain 提议是对以太坊共识层的一次大规模重新设计,旨在实现以太坊的长期愿景和技术目标。以…

moduo之线程池ThreadPool

简介 moduo实现了通用业务线程池&#xff0c;其任务使用无参的函数对象&#xff0c;其任务队列使用的是有界的队列 结构 #mermaid-svg-1tUORFhOtpA7H0hc {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-1tUORFhOtp…