> **摘要**：系统拆解 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

```scala
// 签名：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 条

```scala
// 签名：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）

```scala
// 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 — 写入文本文件

```scala
// 签名：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 — 全局聚合

```scala
// 签名：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 — 灵活的分区内/区间聚合

```scala
// 求平均值
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 — 树形聚合（推荐）

```scala
// 树形聚合避免单点 OOM
rdd.treeReduce(_ + _, depth = 3)       // 推荐用于大数据集
rdd.treeAggregate(zero)(seqOp, combOp, depth = 3)
```

---

## 五、统计类 Action — 便捷统计函数

```scala
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。**

摘要：系统拆解 Spark 全部 Action 算子——按输出类、存储类、聚合类、统计类四大分类，逐个剖析语法、底层原理、Driver/Executor 数据流向和性能陷阱。配有 2 张原创架构图、完整 Scala 代码示例和常见 OOM 排查指南。面向 Java、大数据及 AI 开发工程师。

SparkCore 之 Spark Action 类算子详解

开发工具

Action算子触发Spark计算核心，分输出、存储、聚合、统计四类。collect/take/countByKey数据汇聚Driver易OOM，用take或treeReduce替代。foreachPartition最优写外部系统，saveAsTextFile需预删目录。reduce/aggregate分区聚合防Shuffle压力，配速查表与陷阱排查指南。

代码解释

大数据

数据流

Spark 

Java

HDFS

2026上云采购 | AI焕新·智启新局

bigdata-class

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Action算子触发Spark计算核心，分输出、存储、聚合、统计四类。collect/take/countByKey数据汇聚Driver易OOM，用take或treeReduce替代。foreachPartition最优写外部系统，saveAsTextFile需预删目录。reduce/aggregate分区聚合防Shu...

SparkCore 之 Spark Action 类算子详解-腾讯云开发者社区-腾讯云

SparkCore 之 Spark Action 类算子详解

SparkCore 之 Spark Action 类算子详解

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐