spark-RDD原理

1.缓存和checkpoint机制

1.1缓存的使用

缓存级别

指定缓存的数据位置
默认是缓存到内存上

StorageLevel.DISK_ONLY # 将数据缓存到磁盘上
StorageLevel.DISK_ONLY_2 # 将数据缓存到磁盘上保存两份
StorageLevel.DISK_ONLY_3 # 将数据缓存到磁盘上保存三份
StorageLevel.MEMORY_ONLY # 将数据缓存到内存默认
StorageLevel.MEMORY_ONLY_2 # 将数据缓存到内存保存两份
StorageLevel.MEMORY_AND_DISK # 将数据缓存到内存和磁盘优先将数据缓存到内存上，内存不足可以缓存到磁盘
StorageLevel.MEMORY_AND_DISK_2 = # 将数据缓存到内存和磁盘
StorageLevel.OFF_HEAP # 不使用缓存在系统管理的内存上 heap jvm的java虚拟机中的heap
StorageLevel.MEMORY_AND_DISK_ESER # 将数据缓存到内存和磁盘序列化操作，按照二进制存储，节省空间

使用

persist 使用该方法
cache 内部调用persist
手动释放 unpersist

# 缓存 实现持久化
from pyspark import SparkContext
from pyspark.storagelevel import StorageLevel
sc = SparkContext()# 获取数据
rdd = sc.parallelize([1, 2, 3, 4, 5])
rdd_line = rdd.map(lambda x:x.split(','))
# 将数据转为kv结构
def func(x)return (x[2],int(x[3]))rdd_kv = rdd_line.map(func)# 进入reduce阶段
# 先对kv数据进行分组
rdd_groupby = rdd_kv.groupByKey()
# 对分组后的结果进行缓存
# # storageLevel 修改缓存级别
rdd_groupby.persist(storageLevel=StorageLevel.MEMORY_AND_DISK)
# 触发缓存
rdd_groupby.collect()# 获取kv数据中value部分数据
rdd_count = rdd_groupby.mapValues(lambda x:len(list(x)))# 查看数据
res_count = rdd_count.collect()
print(res_count)

1.2checkpoint

也是将中间rdd数据存储起来，但是存储的位置实时分布式存储系统，可以进行永久保存，程序结束不会释放

# checkpoint 持久化  将数据存储在hdfs上
from pyspark import SparkContext# 创建对象
sc = SparkContext()# 指定checkpoint存储的hdfs位置
sc.setCheckpointDir('hdfs://node1:8020/checkpoint')# 生成rdd数据
rdd = sc.parallelize(['hadoop,spark','spark,python'])# 字符串数据切割
rdd_split = rdd.map(lambda x:x.split(','))  # [[hadoop,spark],[spark,python]]# 将二维列表转为一维
def func(x):return xrdd_word = rdd_split.flatMap(func) # [hadoop,spark,spark,python]
# 持久化操作，可以使用缓存或checkpoint
# # 对rdd使用checkpoint
rdd_word.checkpoint()
# rdd_word.persist()
# # # 触发执行
print(rdd_word.getCheckpointFile())# 将数据转为kv
rdd_kv1 = rdd_word.map(lambda x:(x,1))rdd_kv2 = rdd_word.map(lambda x:(x,2))rdd_kv3 = rdd_word.map(lambda x:(x,3))rdd_kv4 = rdd_word.map(lambda x:(x,4))rdd_kv5 = rdd_word.map(lambda x:(x,5))# 查看kv数据
res = rdd_kv1.collect()
print(res)
res2 = rdd_kv2.collect()
print(res2)
res3 = rdd_kv3.collect()
print(res3)
res4 = rdd_kv4.collect()
print(res4)
res5 = rdd_kv5.collect()
print(res5)

2.数据共享

2.1广播变量

如果要在分布式计算里面分发大的变量数据，这个都会由Driver端进行分发，一般来讲，如果这个变量不是广播变量，那么每个task就会分发一份，这在task数目十分多的情况下Driver的带宽会成为系统的瓶颈，而且会大量消耗task服务器上的资源，如果将这个变量声明为广播变量，那么每个executor拥有一份，这个executor启动的task会共享这个变量，节省了通信的成本和服务器的资源。

减少task线程对应变量的定义，节省内存空间

# 广播变量
from pyspark import SparkContextsc  = SparkContext()num = 10
# 将变量定义成广播变量
b_obj = sc.broadcast(num)rdd = sc.parallelize([1,2,3,4])# 转化计算
def func(x):# 广播变量无法修改# b_obj.value=20# 获取广播变量值return x+b_obj.valuerdd_map = rdd.map(func)# 查看数据
res = rdd_map.collect()
print(res)

2.2累加器

避免资源抢占造成计算错误

# 累加器
from pyspark import SparkContextsc  = SparkContext()num = 10
# 将变量定义成累加器
a_obj = sc.accumulator(num)
# 生成rdd
rdd = sc.parallelize([1,2,3,4])# 对rdd进行计算
def func(x):print(x) # 输出rdd中元素数据# 对累加器的值进行修改 每次加1a_obj.add(1)return (x,1)rdd_map = rdd.map(func)# 查看数据
res = rdd_map.collect()
print(res)# 查看累加器的数据
print(a_obj.value)

3.RDD的依赖关系

窄依赖
- 每个父RDD的一个Partition最多被子RDD的一个Partition所使用
  - map
  - flatMap
  - filter
宽依赖
- 一个父RDD的Partition会被多个子RDD的Partition所使用
  - groupbykey
  - reducebykey
  - sortBykey
- 在宽依赖中rdd之间会发生数据交换，这个交换的过程称为rdd的shuffle
  - 只要是宽依赖必然发生shuffle
  - 在宽依赖进行数据交换时，只有等待所有分区数据交换完成后，才能进行后续的计算，非常影响计算速度

DAG 管理维护rdd之间依赖关系，保证代码的执行顺序。

4.shuffle过程

在 Spark 中，Shuffle 是一种将数据在不同的分区之间重新分布的过程，通常发生在一些特定的操作中，如 groupByKey、reduceByKey、join 等。Shuffle 过程涉及到数据的重新分区、排序和聚合等操作，对 Spark 作业的性能有很大的影响。

以下是 Spark Shuffle 的大致过程：

一、Mapper 阶段（Map 任务）

数据处理：

每个输入分区的数据被分配到一个或多个 Mapper（Map 任务）进行处理。Mapper 会对输入数据进行转换操作，生成中间结果。
例如，在 reduceByKey 操作中，Mapper 会将输入数据中的每个键值对进行处理，生成中间的键值对结果，其中键相同的值会被聚合在一起。

分区函数：

根据指定的分区函数，将中间结果分配到不同的分区中。分区函数决定了每个键值对应该被分配到哪个分区。
例如，在 HashPartitioner（默认的分区函数）中，根据键的哈希值来确定分区。如果有两个分区，键的哈希值对 2 取模，结果为 0 的键值对分配到一个分区，结果为 1 的键值对分配到另一个分区。

缓存中间结果：

Mapper 会将中间结果缓存在内存中，以便在后续的 Shuffle Write 阶段进行写入。如果内存不足，中间结果可能会被溢出到磁盘上。

二、Shuffle Write（混洗写阶段）

数据写入：

Mapper 任务将中间结果写入本地磁盘。每个 Mapper 会根据目标分区的数量，将数据写入多个文件，每个文件对应一个目标分区。
这些文件通常是临时文件，包含了要发送到不同 Reducer 的数据。

索引文件：

同时，Mapper 会为每个输出文件生成一个索引文件，记录了每个分区的数据在输出文件中的偏移量。索引文件用于在 Shuffle Read 阶段快速定位数据。

三、Shuffle Read（混洗读阶段）

数据读取：

Reducer（Reduce 任务）从各个 Mapper 的本地磁盘读取属于自己的分区数据。Reducer 会根据分区的索引文件，确定要读取哪些文件以及从文件中的哪个位置开始读取。
Reducer 会从多个 Mapper 读取数据，然后对相同键的数据进行聚合或其他操作。

数据合并：

Reducer 会将从不同 Mapper 读取的数据进行合并和排序。如果有多个 Mapper 输出了相同键的数据，Reducer 会将这些数据合并在一起，并按照键进行排序。

聚合操作：

最后，Reducer 对合并后的数据进行聚合操作，生成最终的结果。聚合操作可以是求和、计数、求平均值等。

四、性能影响因素和优化

数据倾斜：

如果某些键的值在数据集中出现的频率非常高，可能会导致数据倾斜。这意味着某些 Reducer 会处理大量的数据，而其他 Reducer 处理的数据量很少。数据倾斜会严重影响作业的性能，导致某些任务运行时间过长。
解决数据倾斜的方法包括使用更合适的分区函数、对倾斜的键进行特殊处理（如随机前缀）、在 Mapper 阶段进行预聚合等。

内存管理：

Shuffle 过程中需要大量的内存来缓存中间结果和读取的数据。如果内存不足，可能会导致数据溢出到磁盘上，增加磁盘 I/O 开销，降低性能。
可以通过调整 Spark 的内存参数（如 spark.executor.memory、spark.storage.memoryFraction 等）来优化内存使用。此外，也可以使用一些内存优化技术，如序列化、压缩等，减少内存占用。

网络传输：

Shuffle 过程中需要在不同的节点之间传输大量的数据。网络传输的性能会影响作业的整体执行时间。
可以通过优化网络配置、使用高效的序列化格式（如 Kryo 序列化）、增加网络带宽等方式来提高网络传输性能。

park的shuffle的两个部分
- shuffle wirte 写
- shuffle read 读
- 会进行文件的读写，影响spark的计算速度
spark的shuffle方法类
- 是spark封装好的处理shuffle的方法
- hashshuffle 类
  - 进行的是hash计算
  - spark1.2版本前主要使用，之后引入了sortshuffle
  - spark2.0之后，删除了hashshuffle ，从2.0版本开始使用sortshuffle类
  - 优化的hashshufulle和未优化
- sortshuffle类
  - 排序方式将相同key值数据放在一起
  - sortshuffle类使用时，有两个方法实现shuffle
    - bypass模式版本和普通模式版本
    - bypass模式版本不会排序，会进行hash操作
    - 普通模式版本会排序进行shuffle
  - 可以通过配置指定按照那种模式执行根据task数量决定默认 task数量小于等于200 采用bypass，task数量超过200个则使用普通模式的方法进行shuffle
  - 一个分区对应一个task，所以task数量由分区数决定