Spark分布式计算中Shuffle Read 和 Shuffle Write的职责和区别

news2024/11/23 12:41:49

在 Spark 的分布式计算中，Shuffle Read 和 Shuffle Write 是两个与数据重新分区和分发相关的重要阶段。它们的主要职责和区别如下：

1. Shuffle Write

Shuffle Write 发生在上游的任务执行阶段，其作用是：

分区数据准备：将当前分区的数据根据目标分区键进行分组和组织，生成中间结果文件。
本地磁盘存储：将这些中间结果（通常是分区文件）写入磁盘，供下游任务读取。
核心逻辑：
- 数据被以键值对形式分区。
- 对于每个目标分区，生成一个或多个文件。
- 写文件时使用的机制通常包括 缓冲区写入 和 本地磁盘 I/O。

网络传输：Shuffle Write 通常只涉及本地磁盘写操作，不涉及网络传输。

2. Shuffle Read

Shuffle Read 发生在下游的任务执行阶段，其作用是：

从多个上游节点获取数据：读取上游任务生成的 Shuffle Write 文件，这可能涉及远程网络传输。
重组和反序列化数据：将分区文件中的数据读取到内存中，重新组装为逻辑分区。
核心逻辑：
- 每个下游任务会获取其所需的所有上游分区数据（即所有 Shuffle Write 文件中对应它分区的数据）。
- 数据可能来自本地磁盘，也可能通过网络从远程节点拉取。

网络传输：当下游任务需要的数据分布在其他节点时，Shuffle Read 会涉及网络传输。

3. 两者的区别

属性	Shuffle Write	Shuffle Read
阶段	上游任务输出阶段	下游任务输入阶段
数据位置	写入本地磁盘	读取本地或远程数据
是否涉及网络	不涉及	可能涉及网络传输
主要消耗	磁盘 I/O	网络传输 + 内存解压缩/反序列化
性能优化	数据压缩、批量写文件	数据本地化、优化读取策略

4. 哪些算子会触发 Shuffle？

Shuffle 主要由 宽依赖（Wide Dependency） 的算子触发，如：

groupByKey、reduceByKey：根据键值分组。
join、cogroup：在多个 RDD 之间进行分区重组。
repartition、sortBy：需要对数据重新分区或排序。

5. 性能优化

由于 Shuffle 是分布式计算中性能开销较大的阶段，优化 Shuffle 的重点是减少网络传输和磁盘 I/O：

数据压缩：启用 Shuffle 的压缩（spark.shuffle.compress）。
数据本地化：尽量让任务读取本地数据，减少远程网络读取。
调节并行度：通过调整 spark.sql.shuffle.partitions 控制分区数。
避免不必要的 Shuffle：通过算法改进（如 mapPartitions 替代 groupByKey）。

示例

Shuffle 发生场景

假设我们对一个 RDD 执行 reduceByKey：

val conf = new SparkConf().setAppName("ShuffleExample").setMaster("local[*]")
val sc = new SparkContext(conf)

val rdd = sc.parallelize(Seq(("a", 1), ("b", 2), ("a", 3), ("b", 4)), 2)
val reduced = rdd.reduceByKey(_ + _)
println(reduced.collect().mkString(", "))

Shuffle Write：
- 上游分区会将数据根据键 a 或 b 重组，写入目标分区文件。
- 不同分区可能会产生 ("a", 1) 和 ("b", 2)，分别被写入磁盘。
Shuffle Read：
- 下游任务读取所有目标分区文件，将 ("a", 1) 和 ("a", 3) 拉取到一个分区中，然后计算结果。