Spark rdd算子解析与实践

一、RDD基础回顾

RDD（Resilient Distributed Dataset） 是Spark的核心抽象，代表一个不可变、分区的分布式数据集合。其核心特性包括：

容错性：通过血缘（Lineage）记录数据生成过程，支持丢失分区的自动恢复。
并行计算：数据分片（Partition）存储在集群节点上，并行处理。
惰性求值：转换算子（Transformations）不会立即执行，需触发动作算子（Actions）才会启动计算。

二、RDD算子分类与核心原理

RDD算子分为转换（Transformations）和动作（Actions）两类，其底层依赖关系分为窄依赖（Narrow Dependency）和宽依赖（Wide Dependency）。

算子类型	特点	示例
转换算子	生成新RDD，延迟执行	`map`, `filter`, `groupByKey`
动作算子	触发计算并返回结果到Driver或存储系统	`collect`, `count`, `save`
窄依赖	父RDD的每个分区最多被子RDD的一个分区使用（无需Shuffle）	`map`, `filter`
宽依赖	父RDD的一个分区可能被子RDD的多个分区使用（需Shuffle，性能开销大）	`groupByKey`, `join`

三、常用转换算子详解与示例

1. 单分区操作（Narrow Dependency）

map(func)

功能：对每个元素应用函数，生成新RDD。

示例：将数字列表平方。

val rdd = sc.parallelize(1 to 5)
val squared = rdd.map(x => x * x)  // [1, 4, 9, 16, 25]

filter(func)

功能：筛选满足条件的元素。

示例：过滤偶数。

val filtered = rdd.filter(_ % 2 == 0)  // [2, 4]

flatMap(func)

功能：将每个元素转换为多个输出（展平结果）。

示例：拆分句子为单词。

val lines = sc.parallelize(List("Hello World", "Hi Spark"))
val words = lines.flatMap(_.split(" "))  // ["Hello", "World", "Hi", "Spark"]

2. 键值对操作（Key-Value Pairs）

reduceByKey(func)

功能：按Key聚合，在Shuffle前进行本地Combiner优化。

示例：统计单词频率。

val pairs = words.map(word => (word, 1))
val counts = pairs.reduceByKey(_ + _)  // [("Hello",1), ("World",1), ...]

groupByKey()

功能：按Key分组（无Combiner，性能低于reduceByKey）。

示例：分组后手动统计。

val grouped = pairs.groupByKey()  // [("Hello", [1]), ("World", [1]), ...]
val counts = grouped.mapValues(_.sum)

3. 重分区与Shuffle

repartition(numPartitions)

功能：调整分区数（触发全量Shuffle）。

场景：数据倾斜时增加并行度。

val rdd = sc.parallelize(1 to 100, 2)
val repartitioned = rdd.repartition(4)  // 4个分区

coalesce(numPartitions, shuffle=false)

功能：减少分区数（默认不Shuffle）。

场景：合并小文件写入HDFS。

val coalesced = rdd.coalesce(1)  // 合并为1个分区

四、常用动作算子与实战应用

1. 数据收集与输出

collect()

功能：将RDD所有数据返回到Driver端（慎用大数据集）。
```
val data = rdd.collect()  // Array[Int]
```

saveAsTextFile(path)

功能：将RDD保存为文本文件。

counts.saveAsTextFile("hdfs://path/output")

2. 聚合统计

count()

功能：返回RDD元素总数。
```
val total = rdd.count()  // Long
```

reduce(func)

功能：聚合所有元素（需满足交换律和结合律）。
```
val sum = rdd.reduce(_ + _)  // 15 (1+2+3+4+5)
```

五、高级算子与性能优化

1. Shuffle优化策略

避免groupByKey：优先使用reduceByKey或aggregateByKey（预聚合减少数据传输）。
调整分区数：通过spark.sql.shuffle.partitions控制Shuffle后的分区数量。

2. 持久化与缓存

cache() / persist()：将频繁访问的RDD缓存到内存或磁盘。

val cachedRDD = rdd.cache()  // MEMORY_ONLY
cachedRDD.unpersist()        // 释放缓存

3. Checkpoint机制

作用：切断血缘关系，将RDD持久化到可靠存储（如HDFS）。
```
sc.setCheckpointDir("hdfs://checkpoint")
rdd.checkpoint()
```

六、经典案例：WordCount实现

val textFile = sc.textFile("hdfs://input.txt")
val words = textFile.flatMap(line => line.split(" "))
val pairs = words.map(word => (word, 1))
val counts = pairs.reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://wordcount_output")