4. RDD 算子实战（上）：map、filter、flatMap、mapPartitions、distinct、union

好，咱们今天开始动手。前面几章把 RDD 的血缘、依赖、调度讲了个遍，但说句实话，如果你没亲手写过几个算子，那些概念始终是飘着的。我个人习惯是：先跑通，再优化。所以这一章，咱们把最常用的六个转换算子挨个过一遍。

我会把每个算子的用法、坑、什么时候该用别的替代方案都讲清楚。你跟着敲一遍，Spark 就算入门了。

4.1 map：最基础的“一对一”转换

map 是 Spark 里最常用的算子，没有之一。它的逻辑很简单：对 RDD 里的每一个元素应用一个函数，输出一个新的元素。输入一条，输出一条，一一对应。

// 假设我们有一个用户行为日志 RDD，每条是 "userId,action,timestamp"
val logs = sc.parallelize(Seq(
  "1001,click,2024-01-01 10:00:00",
  "1002,view,2024-01-01 10:01:00",
  "1003,purchase,2024-01-01 10:02:00"
))

// 用 map 提取 userId
val userIds = logs.map(line => line.split(",")(0))
userIds.collect().foreach(println)
// 输出: 1001, 1002, 1003

4.2 filter：只留你想要的

filter 接收一个返回布尔值的函数，保留返回 true 的元素，丢弃 false 的。说白了就是“筛子”。

// 过滤出购买行为
val purchaseLogs = logs.filter(line => line.contains("purchase"))
println(purchaseLogs.count()) // 输出: 1

4.3 flatMap：一对多，展平

flatMap 和 map 很像，区别在于：map 的返回值是一个元素，而 flatMap 的返回值是一个可迭代的集合，并且 Spark 会自动把这个集合“展平”成多个元素。

举个例子：你有一条句子，想把它拆成单词。用 map 你会得到一个 RDD，里面每个元素是一个数组；用 flatMap 你会得到一个 RDD，里面每个元素是一个单词。

val sentences = sc.parallelize(Seq(
  "hello world",
  "spark is awesome",
  "flatMap is useful"
))

// map 的结果：RDD[Array[String]]
val wordsWithMap = sentences.map(_.split(" "))
// 输出: Array(Array("hello","world"), Array("spark","is","awesome"), ...)

// flatMap 的结果：RDD[String]
val words = sentences.flatMap(_.split(" "))
words.collect().foreach(println)
// 输出: hello, world, spark, is, awesome, flatMap, is, useful

4.4 mapPartitions：分区级别的“批处理”

这是很多新手容易忽略的算子。它和 map 的区别在于：map 是对每个元素调用一次函数，而 mapPartitions 是对每个分区调用一次函数，函数接收的是整个分区的迭代器。

为什么要用这个？因为减少函数调用次数。如果你的函数里有创建连接、打开文件、初始化对象等开销大的操作，用 mapPartitions 可以只做一次，而不是每条记录做一次。

// 假设我们要给每条数据打上时间戳
// 用 map 的方式：每条记录都 new 一个 Date 对象
val rddWithMap = rdd.map { record =>
  val date = new java.util.Date()  // 每条记录都创建一次
  (record, date.toString)
}

// 用 mapPartitions 的方式：每个分区只创建一次
val rddWithPartition = rdd.mapPartitions { iter =>
  val date = new java.util.Date()  // 每个分区只创建一次
  iter.map(record => (record, date.toString))
}

对比维度	map	mapPartitions
调用次数	每条记录一次	每个分区一次
输入	单个元素	整个分区的迭代器
输出	单个元素	迭代器（可返回多个元素）
适用场景	简单转换、字段提取	需要初始化资源、批量操作
内存风险	低	高（分区数据可能很大）

4.5 distinct：去重，但代价不小

distinct 用于去除 RDD 中的重复元素。用法很简单，但背后的代价你需要知道。

val numbers = sc.parallelize(Seq(1, 2, 2, 3, 3, 3, 4, 5, 5))
val uniqueNumbers = numbers.distinct()
uniqueNumbers.collect().foreach(println)
// 输出: 1, 2, 3, 4, 5 （顺序可能不同）

我有个建议：如果只是想知道有多少个不重复的值，用 countApproxDistinct 替代，它基于 HyperLogLog 算法，速度快得多，误差在可接受范围内。

4.6 union：合并两个 RDD

union 把两个 RDD 合并成一个。注意，它不会去重，也不会改变分区数。两个 RDD 的分区会简单地拼接在一起。

val rdd1 = sc.parallelize(Seq(1, 2, 3))
val rdd2 = sc.parallelize(Seq(3, 4, 5))
val unionRDD = rdd1.union(rdd2)
unionRDD.collect().foreach(println)
// 输出: 1, 2, 3, 3, 4, 5 （注意有两个 3）

4.7 实战组合：一个完整的例子

咱们把今天学的算子串起来，做一个实际点的例子：处理用户访问日志，统计每个用户的独立访问页面数。

// 模拟日志数据：userId,pageUrl,timestamp
val logs = sc.parallelize(Seq(
  "u001,/home,2024-01-01 10:00",
  "u001,/search,2024-01-01 10:05",
  "u001,/home,2024-01-01 10:10",
  "u002,/home,2024-01-01 10:00",
  "u002,/cart,2024-01-01 10:02",
  "u003,/home,2024-01-01 10:00"
))

// 1. 用 map 提取 (userId, pageUrl) 对
val userPagePairs = logs.map { line =>
  val parts = line.split(",")
  (parts(0), parts(1))
}

// 2. 用 distinct 去重（同一个用户访问同一页面只算一次）
val uniqueVisits = userPagePairs.distinct()

// 3. 用 map 转换成 (userId, 1)
val userOne = uniqueVisits.map { case (user, _) => (user, 1) }

// 4. 用 reduceByKey 统计每个用户的独立页面数
val userPageCount = userOne.reduceByKey(_ + _)

userPageCount.collect().foreach(println)
// 输出: (u001,2), (u002,2), (u003,1)

4.8 本章小结

今天这六个算子，是 Spark 开发者的基本功。我总结一下：

• map：一对一，最常用，但别在里面做重操作。
• filter：筛数据，注意过滤后分区可能不均匀。
• flatMap：一对多，拆词、展平场景首选。
• mapPartitions：分区级批处理，适合初始化资源的场景，但小心 OOM。
• distinct：去重，代价是 Shuffle，数据量大时考虑近似算法。
• union：合并 RDD，注意分区数会叠加。