摘要：系统拆解 Spark 全部 Action 算子——按输出类、存储类、聚合类、统计类四大分类，逐个剖析语法、底层原理、Driver/Executor 数据流向和性能陷阱。配有 2 张原创架构图、完整 Scala 代码示例和常见 OOM 排查指南。面向 Java、大数据及 AI 开发工程师。
一、Action vs Transformation — 本质区别
Action（行动算子）是 Spark 中触发计算的唯一入口。所有 Transformation 只构建 DAG，Action 才是真正按下"执行按钮"的那个操作。
| 维度 | Transformation | Action |
|---|---|---|
| 返回值 | 新 RDD | 非 RDD（值/数组/写入存储） |
| 触发计算 | ❌ 不触发 | ✅ 触发 DAG 执行 |
| DAGScheduler | 不参与 | 触发 Stage 划分 + Task 提交 |
| 执行时机 | 惰性求值 | 立即触发 |
| 代表 | map / filter / join | count / collect / saveAsTextFile |
二、输出类 Action — 将数据拉回 Driver
2.1 collect — 收集全部数据到 Driver
// 签名：def collect(): Array[T]
// ⚠️ 所有数据通过网络传输到 Driver 内存
// 数据量大时 → Driver OOM！
val rdd = sc.parallelize(1 to 1000)
rdd.collect() // Array[Int](1,2,...,1000) — 全部在Driver内存中
// ❌ 危险用法：PB 级数据 collect
sc.textFile("hdfs://100TB-logs/").collect() // Driver OOM 必现！
// ✅ 正确用法：数据量小（<几MB）时才用 collect
rdd.filter(_.contains("specific_keyword")).collect() // 过滤后数据量小
2.2 take — 取前 N 条
// 签名：def take(num: Int): Array[T]
// 只拉指定数量到 Driver，不会 OOM
rdd.take(10) // 前10条
rdd.take(100) // 前100条
// take 的底层执行：
// 先取第一个 Partition → 不够再取第二个 → 直到凑满 num 条
// 性能优于 collect（尤其数据量大时）
2.3 foreach / foreachPartition — 遍历执行（不返回Driver）
// foreachPartition: 每个分区执行一次（推荐！）
rdd.foreachPartition { iter =>
val conn = DriverManager.getConnection(url) // 每个分区创建1次连接
iter.foreach { record =>
conn.execute(s"INSERT INTO t VALUES ($record)")
}
conn.close()
}
// mapPartitions + foreachPartition 组合是写入外部系统的最优模式
三、存储类 Action — 将数据写入外部存储
3.1 saveAsTextFile — 写入文本文件
// 签名：def saveAsTextFile(path: String)
// 每个 Partition 写入一个文件（part-00000, part-00001, ...）
rdd.saveAsTextFile("hdfs://output/result/")
// ⚠️ 目标目录不能已存在（否则抛异常）
// 解决方案：先删除目录
val path = new Path("hdfs://output/result/")
path.getFileSystem(sc.hadoopConfiguration).delete(path, true)
rdd.saveAsTextFile("hdfs://output/result/")
四、聚合类 Action — 在 Executor 聚合后返回 Driver
4.1 reduce — 全局聚合
// 签名：def reduce(f: (T, T) => T): T
// 先在每个分区内聚合 → Shuffle → 最终聚合 → 返回Driver
val rdd = sc.parallelize(1 to 100)
rdd.reduce(_ + _) // 5050
rdd.reduce(_ max _) // 100
// ⚠️ reduce 要求函数满足结合律和交换律
4.2 aggregate — 灵活的分区内/区间聚合
// 求平均值
val rdd = sc.parallelize(1 to 100)
val (sum, count) = rdd.aggregate((0, 0))(
seqOp = { case ((s, c), v) => (s + v, c + 1) },
combOp = { case ((s1, c1), (s2, c2)) => (s1 + s2, c1 + c2) }
)
val avg = sum.toDouble / count // 50.5
4.3 treeAggregate / treeReduce — 树形聚合（推荐）
// 树形聚合避免单点 OOM
rdd.treeReduce(_ + _, depth = 3) // 推荐用于大数据集
rdd.treeAggregate(zero)(seqOp, combOp, depth = 3)
五、统计类 Action — 便捷统计函数
rdd.count() // 计数
rdd.countByKey() // 按Key统计 → Map[K, Long] (⚠️ OOM)
rdd.max() // 最大值
rdd.min() // 最小值
rdd.sum() // 求和
rdd.histogram(10) // 直方图
rdd.toDebugString // 查看Lineage
六、Action 算子速查表
| 算子 | 分类 | 返回 | 数据流向 | OOM 风险 |
|---|---|---|---|---|
collect() |
输出 | Array[T] | →Driver | ⚠️ 高 |
take(n) |
输出 | Array[T] | →Driver | ✅ 低 |
foreachPartition |
输出 | Unit | →Executor | ✅ 无 |
saveAsTextFile |
存储 | Unit | →磁盘 | ✅ 无 |
reduce(f) |
聚合 | T | →Driver | ✅ 低 |
aggregate |
聚合 | U | →Driver | ✅ 低 |
treeReduce |
聚合 | T | →Driver | ✅ 极低 |
count() |
统计 | Long | →Driver | ✅ 极低 |
countByKey() |
统计 | Map | →Driver | ⚠️ 高 |
七、常见 Action 陷阱与排查
| 陷阱 | 现象 | 解决 |
|---|---|---|
| collect OOM | OutOfMemoryError |
take(n) 替代 |
| countByKey OOM | Driver GC overhead | reduceByKey(_+_).collect() |
| foreach 打印不显示 | 输出在Executor日志 | collect().foreach(println) |
| saveAsTextFile 目录已存在 | FileAlreadyExistsException |
先删除目录 |
写在最后
Action 算子虽然数量不多（约 20 个），但理解它们的数据流向和内存压力至关重要：
- collect/countByKey → Driver：数据向 Driver 汇聚 → 控制数据量
- foreach/saveAsTextFile → Executor/存储：数据向外发散 → 无内存压力
- reduce/aggregate → 聚合后返回：Executor 间聚合 → 关注 Shuffle
选对 Action，不仅决定性能，更决定你的 Driver 会不会 OOM。
