首页 1 2 3 4 5 6 7

Spark大数据处理讲课笔记3.3 掌握RDD分区

文章目录
零、本讲学习目标
一、RRD分区
（一）RDD分区概念
（二）RDD分区作用
二、RDD分区数量
（一）RDD分区原则
（二）影响分区的因素
（三）使用parallelize()方法创建RDD时的分区数量
1、指定分区数量
2、默认分区数量
3、分区源码分析
（四）RDD分区方式
（五）使用textFile()方法创建RDD时的分区数量
1、指定最小分区数量
2、默认最小分区数量
3、默认实际分区数量
三、Spark分区器
（一）分区器 - Partitioner抽象类
（二）哈希分区器 - HashPartitioner类
四、自定义分区器
（一）提出问题
（二）解决问题
1、新建自定义分区器
2、使用自定义分区器
3、项目打包上传服务器
4、提交程序运行
零、本讲学习目标
学会如何指定分区数量
会定义与使用自定义分区器
一、RRD分区
（一）RDD分区概念

RDD是一个大的数据集合，该集合被划分成多个子集合分布到了不同的节点上，而每一个子集合就称为分区（Partition）。因此，也可以说，RDD是由若干个分区组成的。

（二）RDD分区作用
在分布式程序中，网络通信的开销是很大的，因此控制数据分布以获得最少的网络传输可以极大的提升程序的整体性能，Spark程序可以通过控制RDD分区方式来减少通信开销。Spark中所有的RDD都可以进行分区，系统会根据一个针对键的函数对元素进行分区。虽然Spark不能控制每个键具体划分到哪个节点上，但是可以确保相同的键出现在同一个分区上。
二、RDD分区数量
（一）RDD分区原则
RDD各个分区中的数据可以并行计算，因此分区的数量决定了并行计算的粒度。Spark会给每一个分区分配一个单独的Task任务对其进行计算，因此并行Task的数量是由分区的数量决定的。RDD分区的一个分区原则是使得分区的数量尽量等于集群中CPU核心数量。
（二）影响分区的因素
RDD的创建有两种方式：一种是使用parallelize()方法从对象集合创建；另一种是使用textFile()方法从外部存储系统创建。而RDD分区的数量与RDD的创建方式以及Spark集群的运行模式有关。
（三）使用parallelize()方法创建RDD时的分区数量
1、指定分区数量
使用parallelize()方法创建RDD时，可以传入第二个参数，指定分区数量。

分区的数量应尽量等于集群中所有CPU的核心总数，以便可以最大程度发挥CPU的性能。
利用mapPartitionsWithIndex()函数实现带分区索引的映射

2、默认分区数量
若不指定分区数量，则默认分区数量为Spark配置文件spark-defaults.conf中的参数spark.default.parallelism的值。若没有配置该参数，则Spark会根据集群的运行模式自动确定分区数量。
如果是本地模式，默认分区数量就等于本机CPU核心总数，这样每个CPU核心处理一个分区的计算任务，可以最大程度发挥CPU的性能。

如果是Spark Standalone或Spark On YARN模式，默认分区数量就取集群中所有CPU的核心总数与2中的较大值，即最少分区数为2。

3、分区源码分析
parallelize()方法是在SparkContext类定义的
（四）RDD分区方式
Spark框架为RDD提供了两种分区方式，分别是哈希分区（HashPartitioner）和范围分区（RangePartitioner）。其中，哈希分区是根据哈希值进行分区；范围分区是将一定范围的数据映射到一个分区中。这两种分区方式已经可以满足大多数应用场景的需求。与此同时，Spark也支持自定义分区方式，即通过一个自定义的Partitioner对象来控制RDD的分区，从而进一步减少通信开销。
（五）使用textFile()方法创建RDD时的分区数量
textFile()方法通常用于读取HDFS中的文本文件，使用该方法创建RDD时，Spark会对文件进行分片操作（类似于MapReduce的分片，实际上调用的是MapReduce的分片接口），分片操作完成后，每个分区将存储一个分片的数据，因此分区的数量等于分片的数量。

2、默认最小分区数量

若不指定最小分区数量，则Spark将采用默认规则计算默认最小分区数量。

2、新建科目分区器
创建net.huawei.rdd.day04包，在包里创建SubjectPartitioner类

package net.huawei.rdd.day04

import org.apache.spark.Partitioner

class SubjectPartitioner(partitions: Int) extends Partitioner {

override def numPartitions: Int = partitions

override def getPartition(key: Any): Int = {
val partitionIndex = key.toString match {
case "chinese" => 0
case "math" => 1
case "english" => 2
}
partitionIndex
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
3、测试科目分区器
调用RDD的partitionBy()方法传入科目分区器类SubjectPartitioner的实例，可以对RDD按照自定义规则进行重新分区。
在net.huawei.rdd.day04包里创建TestSubjectPartitioner单例对象

package net.huawei.rdd.day04

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object TestSubjectPartitioner {
def main(args: Array[String]): Unit = {
// 创建Spark配置对象
val conf = new SparkConf()
.setAppName("TestSubjectPartitioner") // 设置应用名称
.setMaster("local[*]") // 设置主节点位置（本地调试）
// 基于Spark配置对象创建Spark容器
val sc = new SparkContext(conf)
// 读取HDFS文件，生成RDD
val lines = sc.textFile("hdfs://master:9000/partition/input/marks.txt")
// 将每行数据映射成(科目,成绩)二元组
val data: RDD[(String, Int)] = lines.map(line => {
val fields = line.split(" ")
(fields(0), fields(1).toInt) // (科目,成绩)
})
// 将数据按科目分区器重新分区
val partitionData = data.partitionBy(new SubjectPartitioner(3))
// 在控制台输出分区数据
partitionData.collect.foreach(println)
// 保存分区数据到HDFS指定目录
partitionData.saveAsTextFile("hdfs://master:9000/partition/output")
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
运行程序，查看结果

查看HDFS的结果文件

如果传入的分区数不是3，会出现什么状况？
删除输出目录

运行程序，查看控制台输出结果

查看HDFS上的结果文件

删除输出目录，修改分区数为2，再运行程序，查看控制台结果