当前位置：首页 > ds >正文

RDD 两类操作详解(Scala)：转换与行动

ds 2025/5/15 7:30:16

1. 转换操作（Transformations）

定义

作用：从一个已有的 RDD 生成一个新的 RDD。
特点：延迟执行（Lazy Evaluation），仅记录操作逻辑，不立即计算。

示例：

val rdd1 = sc.parallelize(1 to 10)  
val rdd2 = rdd1.map(_ * 2)  // 转换操作：每个元素乘2  
val rdd3 = rdd2.filter(_ > 10) // 转换操作：过滤大于10的元素

常见转换操作

操作	说明	示例
`map(func)`	对每个元素应用函数	`rdd.map(x => x + 1)`
`filter(func)`	过滤满足条件的元素	`rdd.filter(x => x % 2 == 0)`
`flatMap(func)`	先映射后扁平化（如拆分文本行）	`rdd.flatMap(_.split(" "))`
`groupByKey()`	按键分组（仅适用于 Pair RDD）	`rdd.groupByKey()`
`reduceByKey(func)`	按键聚合（优化版 groupBy + reduce）	`rdd.reduceByKey(_ + _)`

2. 行动操作（Actions）

定义

作用：触发实际计算，返回结果到驱动程序或写入外部存储。
特点：立即执行（Eager Evaluation），会生成 Job 并提交到集群。

示例：

val count = rdd3.count()  // 行动操作：统计元素数量  
rdd3.saveAsTextFile("hdfs://output") // 行动操作：保存结果

常见行动操作

操作	说明	示例
`count()`	返回 RDD 中元素的总数	`rdd.count()`
`collect()`	将所有元素以数组形式返回驱动端	`rdd.collect()`
`take(n)`	返回前 n 个元素	`rdd.take(5)`
`first()`	返回第一个元素（等价于 take(1))	`rdd.first()`
`saveAsTextFile(path)`	将 RDD 保存为文本文件	`rdd.saveAsTextFile("hdfs://path")`
`foreach(func)`	对每个元素应用函数（无返回值）	`rdd.foreach(println)`

两类操作的核心区别

特征	转换操作（Transformations）	行动操作（Actions）
执行时机	延迟执行，记录操作但不触发计算	立即执行，触发实际计算
返回值	返回一个新的 RDD	返回非 RDD 类型（如数值、数组等）
依赖关系	生成 RDD 的血缘关系（Lineage）	不生成新的 RDD，直接输出结果
优化机制	可被 Spark 优化器合并（如流水线执行）	无法优化，直接提交 Job
数据持久化	不缓存数据（除非显式调用 `persist()`）	可能触发缓存（如多次调用时）
用途	定义数据处理流程	获取结果或输出到外部系统

图解执行流程

关键原理：延迟执行（Lazy Evaluation）

优势：
1. 优化执行计划：Spark 将多个转换操作合并为单个 Stage，减少计算步骤。
2. 避免中间结果存储：仅在必要时计算，节省内存和磁盘 I/O。
3. 容错简化：通过血缘关系（Lineage）重建数据，无需保存中间状态。

示例：

val rdd = sc.textFile("data.txt")  .flatMap(_.split(" "))  // 转换1  .map(_.toUpperCase())   // 转换2  .filter(_.length > 3)   // 转换3  // 此时未执行任何计算，仅记录血缘关系  
rdd.count()  // 行动操作触发所有转换的执行

常见误区与注意事项

重复计算问题：
- 若多次调用行动操作，每次都会重新计算整个血缘链。
- 解决方案：对重复使用的 RDD 调用 persist() 或 cache() 持久化。
Shuffle 操作：
- 宽依赖转换（如 groupByKey）会触发 Shuffle，性能开销大。
- 优化建议：优先使用 reduceByKey 代替 groupByKey。
行动操作的数据量限制：
- collect() 会将所有数据拉取到驱动端，可能导致 OOM。
- 替代方案：使用 take(n) 或分批处理。