Spark 行动算子（Action）全面解析

摘要：本文系统梳理 Spark 中的行动算子，涵盖触发机制、常用 API 分类、执行原理与实际使用场景，帮助你真正理解 Action 背后发生了什么。

---

一、为什么要讲行动算子？

在 Spark 的编程模型中，算子分为两大类：

类型	特点	是否触发计算
转换算子（Transformation）	惰性求值，返回新 RDD	❌ 不触发
行动算子（Action）	触发 DAG 执行，返回结果或写出数据	✅ 触发

核心理解：每一次调用 Action，Spark 就会向 Driver 提交一个 Job，Driver 将 DAG 切分成若干 Stage，Stage 内部拆分为 Task 并分发到各 Executor 执行。

调用 Action
    └─► 提交 Job
          └─► DAG Scheduler 切分 Stage
                └─► Task Scheduler 分发 Task
                      └─► Executor 执行 → 结果返回 Driver

---

二、行动算子分类总览

Action 算子
├── 聚合类       collect / count / countByKey / countByValue
├── 取值类       first / take / takeOrdered / takeSample
├── 归约类       reduce / fold / aggregate
├── 遍历类       foreach / foreachPartition
├── 保存类       saveAsTextFile / saveAsObjectFile / saveAsSequenceFile
└── 统计类       max / min / sum / mean / variance / stdev

---

三、常用行动算子详解

3.1 collect()

将 RDD 中所有数据收集到 Driver 端，返回 Array。

val rdd = sc.parallelize(List(1, 2, 3, 4, 5))
val result = rdd.collect()
// result: Array[Int] = Array(1, 2, 3, 4, 5)

⚠️ 注意：生产环境中慎用！数据量过大会导致 Driver OOM。仅适合数据量可控的场景（如测试、调试）。

---

3.2 count()

返回 RDD 中元素的总个数。

val rdd = sc.parallelize(List("a", "b", "c", "a"))
println(rdd.count())  // 4

---

3.3 countByKey()

仅适用于 PairRDD (K, V)，统计每个 Key 出现的次数，返回 Map[K, Long]。

val rdd = sc.parallelize(List(("a", 1), ("b", 2), ("a", 3)))
val result = rdd.countByKey()
// result: Map(a -> 2, b -> 1)

---

3.4 countByValue()

统计 RDD 中每个元素出现的次数，返回 Map[T, Long]。

val rdd = sc.parallelize(List("apple", "banana", "apple", "orange"))
val result = rdd.countByValue()
// result: Map(apple -> 2, banana -> 1, orange -> 1)

---

3.5 first()

返回 RDD 中的第一个元素，等价于 take(1)(0)。

val rdd = sc.parallelize(List(10, 20, 30))
println(rdd.first())  // 10

---

3.6 take(n)

返回 RDD 前 n 个元素组成的数组，不保证顺序（按分区顺序扫描）。

val rdd = sc.parallelize(List(5, 3, 1, 4, 2))
rdd.take(3)  // Array(5, 3, 1)

---

3.7 takeOrdered(n)

返回 RDD 中最小的 n 个元素（升序），可自定义排序规则。

val rdd = sc.parallelize(List(5, 3, 1, 4, 2))
rdd.takeOrdered(3)                        // Array(1, 2, 3)
rdd.takeOrdered(3)(Ordering[Int].reverse) // Array(5, 4, 3) 降序

与 take 的区别：takeOrdered 会在每个 Partition 局部排序后再归并，效率优于 sortBy + take。

---

3.8 reduce(func)

通过一个二元函数对 RDD 所有元素进行归约，要求函数满足交换律和结合律。

val rdd = sc.parallelize(List(1, 2, 3, 4, 5))
val sum = rdd.reduce((a, b) => a + b)  // 15
val max = rdd.reduce((a, b) => if (a > b) a else b)  // 5

执行过程：先在每个 Partition 内部归约，再将各 Partition 结果汇总到 Driver 做最终归约。

---

3.9 aggregate(zeroValue)(seqOp, combOp)

aggregate 是最通用的归约算子，允许返回值类型与输入类型不同。

• seqOp：分区内的聚合函数 (U, T) => U
• combOp：分区间的合并函数 (U, U) => U

// 同时计算总和与元素个数，从而得到平均值
val rdd = sc.parallelize(List(1, 2, 3, 4, 5), 2)
val (sum, count) = rdd.aggregate((0, 0))(
  (acc, num) => (acc._1 + num, acc._2 + 1),   // seqOp
  (a, b) => (a._1 + b._1, a._2 + b._2)         // combOp
)
val avg = sum.toDouble / count  // 3.0

---

3.10 foreach(func)

对 RDD 每个元素执行函数，在 Executor 端执行，不返回值。常用于写入外部系统。

rdd.foreach(x => println(x))  // 输出在 Executor 端，Driver 不可见

---

3.11 foreachPartition(func)

以分区为单位执行函数，每个分区调用一次。适合需要建立连接的场景（如数据库写入），避免每条数据都创建连接。

rdd.foreachPartition { iter =>
  val conn = createDBConnection()  // 每个分区只建一次连接
  iter.foreach { record =>
    conn.write(record)
  }
  conn.close()
}

最佳实践：凡是涉及外部资源（数据库、消息队列、缓存），优先用 foreachPartition 而非 foreach，显著降低连接开销。

---

3.12 saveAsTextFile(path)

将 RDD 保存为文本文件，每个元素调用 toString 写为一行。分区数决定输出文件数。

rdd.saveAsTextFile("hdfs://namenode/output/result")
// 输出: /output/result/part-00000, part-00001, ...

---

四、行动算子执行原理深入

4.1 宽依赖与 Stage 划分

rdd1 ──map──► rdd2 ──filter──► rdd3 ──reduceByKey──► rdd4 ──collect()
                                          ▲
                                      Shuffle 边界
                            ◄── Stage 0 ──┤──── Stage 1 ───►

• map、filter 是窄依赖，同属一个 Stage
• reduceByKey 触发 Shuffle，产生 Stage 边界
• 调用 collect() 触发整个 Job 执行

4.2 Action 与 Job 的关系

val rdd = sc.textFile("data.txt")
  .flatMap(_.split(" "))
  .map((_, 1))
  .reduceByKey(_ + _)

rdd.count()   // Job 1
rdd.collect() // Job 2，DAG 重新执行（除非 cache）

每次 Action 都是一个独立 Job。如果 RDD 会被多次使用，务必调用 cache() 或 persist() 避免重复计算。

---

五、常见误区与最佳实践

❌ 误区 1：在循环中多次调用 Action

// 错误：每次循环都触发一个 Job
for (i <- 1 to 10) {
  println(rdd.count())  // 触发 10 个 Job！
}

// 正确：缓存后复用
val cached = rdd.cache()
val total = cached.count()

❌ 误区 2：用 collect() 处理大数据集

// 危险：数据全量拉到 Driver
rdd.collect().foreach(process)

// 安全：在 Executor 端处理
rdd.foreach(process)
// 或写出到存储
rdd.saveAsTextFile(outputPath)

✅ 最佳实践总结

场景	推荐算子
调试/验证少量数据	take(n)、first()
统计元素数量	count()
全局聚合（同类型）	reduce()
全局聚合（跨类型）	aggregate()
写入外部存储	foreachPartition()
落地到 HDFS	saveAsTextFile()
RDD 多次复用	先 cache()，再 Action

---

六、总结

行动算子是 Spark 程序的"触发器"，理解它的核心在于：

1. 惰性求值：Transformation 只是构建 DAG，Action 才真正触发计算
2. Job 粒度：每个 Action 对应一个 Job，合理减少 Action 调用次数
3. 数据位置：collect/take 将数据拉回 Driver，foreach/save 在 Executor 端处理
4. 缓存策略：多次复用同一 RDD 时，配合 cache() 避免重算

掌握行动算子的选择与优化，是写出高性能 Spark 程序的基础。

---

如有问题欢迎在评论区交流，也欢迎关注后续 Spark 系列文章 🚀