项目3：查询和统计员工薪资数据——Spark Shell编程

先修基础：项目1（Spark概述与集群搭建）、项目2（Scala基础）、Hadoop

第一部分：项目背景与 Spark Shell 入门
第二部分：创建 RDD 的三种方式
第三部分：RDD 基础操作（转换 + 行动）
第四部分：RDD 集合操作
第五部分：键值对 RDD 操作（重点）
第六部分：RDD 连接操作
第七部分：RDD 文件读写
第八部分：项目实战——员工薪资数据查询统计
第九部分：常见问题与排错指南
第十部分：实习 / 面试高频考点
附录：习题解析

第一部分：项目背景与 Spark Shell 入门

1.1 项目背景

为什么要学 Spark Shell 编程？

薪酬调整和绩效管理在现代企业的发展中发挥着重要的作用。一个企业的薪酬体系建立后，通常会不断完善。科学合理的企业薪酬体系能够提升员工的工作热情和积极性，激励他们为企业和社会做出更大贡献。

本项目以员工薪资数据查询统计为实战场景，学习如何使用 Spark Shell 进行 RDD 编程，包括：

创建 RDD
RDD 基础转换操作和行动操作
键值对 RDD 操作
RDD 连接操作
不同格式文件的读写

项目数据说明

公司有员工 2023 年上半年和下半年薪资文件，两份文件格式相同，各 10 个字段：

字段序号	字段名	含义	示例
第1列	EmpID	员工编号	20438
第2列	Name	员工姓名	Deana Lucero
第3列	Gender	性别	F/M
第4列	Date_of_Birth	出生日期	2/11/96
第5列	Age	年龄	23
第6列	GROSS	应发工资	30171
第7列	Net_Pay	实际薪资（实发工资）	27825
第8列	Deduction	扣款	2346
第9列	Designation	职位	Technical Solutions Engineer
第10列	Department	部门	AmaTec - APJ TSE

⚠️ 重点关注：本项目主要用到第2列（姓名）和第7列（实际薪资Net_Pay），注意列索引从0开始，所以是 data(1) 和 data(6)。

1.2 Spark Shell 是什么？

Spark Shell 是 Spark 提供的交互式命令行工具，支持 Scala 和 Python 两种语言（本课程用 Scala）。

特点

交互式：写一行执行一行，边写边看结果，适合学习和调试
自动创建 SparkContext：启动后自动创建 sc 对象，不用手动 new
即写即跑：不用编译打包，直接在命令行写代码执行

启动方式

bash

# YARN Client 模式
spark-shell --master yarn --deploy-mode client

# Standalone 模式
spark-shell --master spark://master:7077

# 本地模式
spark-shell

💡 记忆技巧：spark-shell 启动后看到 scala> 提示符就成功了，sc 变量已经自动创建好了，直接用就行。

第二部分：创建 RDD 的三种方式

RDD（弹性分布式数据集）是 Spark 最核心的数据抽象。创建 RDD 有三种方法：

创建方式	方法	适用场景
从内存集合创建	`parallelize()`、`makeRDD()`	测试、学习、小数据量
从外部存储读取	`textFile()` 等	生产环境、大数据量
从已有RDD转换	各种转换算子	数据处理过程中

2.1 方式一：从内存集合创建 RDD

① parallelize() 方法

最常用的创建方式，接收一个 Seq 集合（List、Array 等），可选指定分区数。

语法：

scala

sc.parallelize(集合, 分区数)

示例代码：

scala

// 定义一个Seq集合
val seq = Seq(1, 2, 3, 4, 5)

// 使用parallelize()方法创建RDD（不指定分区数，默认=CPU核心数）
val rdd = sc.parallelize(seq)

// 查看RDD默认分区个数
rdd.partitions.size

// 设置分区个数为4后创建RDD
val rdd = sc.parallelize(seq, 4)

// 再次查看RDD分区个数
rdd.partitions.size

💡 分区数说明：
不指定分区数时，默认等于程序分配到的 CPU 核心数
分区数决定了并行度，分区越多，并行度越高
但分区不是越多越好，太多会增加调度开销

② makeRDD() 方法

makeRDD() 有两种使用方式：

第一种：和 parallelize() 完全一样，接收集合创建 RDD

scala

val rdd = sc.makeRDD(List(1, 2, 3, 4, 5))

第二种：接收 Seq[(T, Seq[String])] 类型，创建带位置信息的 RDD

T 是 RDD 中存储的值
Seq[String] 是该分区的首选位置（节点列表）
分区个数 = Seq 的个数

示例代码：

scala

// 定义一个带位置信息的Seq集合
val seq = Seq(
  (1, Seq("iteblog.com", "sparkhost1.com")),
  (2, Seq("itebolg.com", "sparkhost2.com")),
  (3, Seq("iteblog.com", "sparkhost3.com"))
)

// 使用makeRDD()方法创建RDD
val rdd = sc.makeRDD(seq)

// 查看RDD的值
rdd.collect()

// 查看分区个数（=3，因为seq有3个元素）
rdd.partitions.size

// 根据位置信息查看每一个分区的首选位置
rdd.preferredLocations(rdd.partitions(0))
rdd.preferredLocations(rdd.partitions(1))
rdd.preferredLocations(rdd.partitions(2))

💡 理解：makeRDD 的第二种方式可以指定数据"倾向于"在哪个节点上计算，实现移动计算不如移动数据的优化。普通开发用得不多，了解即可。

parallelize vs makeRDD 对比

对比项	parallelize()	makeRDD()
基本用法	接收集合创建RDD	接收集合创建RDD（和parallelize一样）
位置信息	不支持	支持指定首选位置（第二种用法）
指定分区数	可以	不能直接指定（由Seq个数决定）
使用频率	⭐⭐⭐⭐⭐ 最常用	⭐⭐ 了解即可

2.2 方式二：从外部存储读取创建 RDD

通过 SparkContext.textFile() 方法读取外部数据创建 RDD，这是生产环境最常用的方式。

支持的数据源

本地文件系统（Linux/Windows）
HDFS（最常用）
压缩文件
目录（读取目录下所有文件）
通配符匹配的文件（如 *.log）

① 读取本地文件

scala

// 读取Linux本地文件test.txt
val local_data = sc.textFile("file:///root/data/bigdata.txt")

// 统计test.txt文件的数据行数
local_data.count()

⚠️ 注意：本地文件路径要加 file:// 前缀，而且每个节点上都要有这个文件才行（或者只在本地模式用）。

② 读取 HDFS 文件（最常用）

scala

hdfs dfs -put /root/data/bigdata.txt /dsj/data/

// 读取HDFS文件test.txt
val hdfs_data = sc.textFile("/dsj/data/bigdata.txt")

// 统计test.txt文件的数据行数
hdfs_data.count()

💡 说明：
HDFS 路径可以省略 hdfs://master:8020 前缀，直接写路径即可
textFile() 每行是 RDD 的一个元素，类型是 String
可以指定分区数：sc.textFile(path, 分区数)

2.3 方式三：从已有 RDD 转换得到新 RDD

通过转换算子（map、filter、flatMap 等）从已有 RDD 生成新的 RDD。这是日常开发中最常见的方式，后续章节会详细讲。

scala

val rdd1 = sc.parallelize(List(1, 2, 3))
val rdd2 = rdd1.map(x => x * 2)  // 转换得到新RDD
val rdd3 = rdd2.filter(_ > 3)    // 再转换得到新RDD

第三部分：RDD 基础操作（转换 + 行动）

RDD 的操作分为两大类：

转换操作（Transformation）：懒执行，返回新 RDD，不触发计算
行动操作（Action）：触发计算，返回结果或写入外部存储

💡 复习：项目1讲过，转换算子就像"列购物清单"，行动算子就像"出门购物"。

3.1 转换算子（Transformation）

① map() 方法—— 一一映射

作用：对 RDD 中的每一个元素应用一个函数，返回新的 RDD。特点：输入多少个元素，输出还是多少个元素（一对一）。类型：转换操作（懒执行）

语法：

scala

rdd.map(func)

示例代码：

scala

// 创建RDD
val rdd = sc.parallelize(List(1, 3, 45, 3, 76))

// map()方法求每个元素的平方值
val sq_rdd = rdd.map(x => x * x)

// 查看结果（行动算子触发计算）
sq_rdd.collect()
// 结果：Array(1, 9, 2025, 9, 5776)

💡 理解：map 就像"流水线"，每个元素经过函数处理后变成另一个元素，数量不变。

② flatMap() 方法—— 先映射再扁平化

作用：先对每个元素应用函数（返回的是集合/数组），再把所有集合的元素"摊平"成一个级别。特点：输入 N 个元素，输出可能多于 N 个（一对多）。 典型应用：切分单词。

语法：

scala

rdd.flatMap(func)

示例代码：

scala

// 创建RDD
val rdd = sc.parallelize(List("How are you", "I am fine", "What about you"))

// 查看原始RDD
rdd.collect()
// 结果：Array("How are you", "I am fine", "What about you")

// 使用map分割字符串（每个元素变成数组）
rdd.map(x => x.split(" ")).collect()
// 结果：Array(Array("How", "are", "you"), Array("I", "am", "fine"), ...)
// 注意：是数组的数组，二维结构

// 使用flatMap分割字符串（扁平化）
rdd.flatMap(x => x.split(" ")).collect()
// 结果：Array("How", "are", "you", "I", "am", "fine", "What", "about", "you")
// 注意：全部摊平成一维了

💡 记忆口诀：flatMap = map + flatten（先map再扁平化）
map vs flatMap 对比：
算子输入输出元素数量
map 3行字符串 3个数组不变
flatMap 3行字符串 9个单词变多（一对多）

算子	输入	输出	元素数量
map	3行字符串	3个数组	不变
flatMap	3行字符串	9个单词	变多（一对多）

③ filter() 方法—— 过滤

作用：根据条件过滤 RDD 中的元素，保留满足条件（返回 true）的元素。特点：输入 N 个元素，输出 ≤ N 个元素。类型：转换操作

语法：

scala

rdd.filter(func)  // func返回Boolean

示例代码：

scala

// 创建RDD
val rdd = sc.parallelize(List(('a', 1), ('b', 2), ('c', 3)))

// 过滤掉第二个值小于等于1的元素
rdd.filter(_._2 > 1).collect()
// 等价于
rdd.filter(x => x._2 > 1).collect()
// 结果：Array(('b', 2), ('c', 3))

💡 说明：_._2 > 1 是 Scala 的简写，_ 代表每个元素，._2 取元组的第二个值。

④ distinct() 方法—— 去重

作用：去除 RDD 中完全相同的重复元素。特点：无参数，宽依赖（会触发 Shuffle）。类型：转换操作

示例代码：

scala

// 创建带重复数据的RDD
val rdd = sc.makeRDD(List(('a', 1), ('a', 1), ('b', 1), ('c', 1)))

// 去重
rdd.distinct().collect()
// 结果：Array(('a', 1), ('b', 1), ('c', 1))

⑤ sortBy() 方法—— 排序

作用：对 RDD 中的元素按指定字段排序。

三个参数：

参数	含义	说明
f: (T) => K	排序依据的函数	必须参数，指定按什么排序
ascending	升序/降序	默认 true（升序），false 为降序
numPartitions	排序后的分区数	默认和排序前相同

示例代码：

scala

// 创建RDD
val rdd = sc.parallelize(List((1, 3), (45, 3), (7, 6)))

// 对元组的第二个值进行降序排序，分区个数设置为1
val sort_rdd = rdd.sortBy(x => x._2, false, 1)

// 查看结果
sort_rdd.collect()
// 结果：Array((7, 6), (1, 3), (45, 3))

💡 说明：
第一个参数必须传，指定按什么排序
第二个参数默认 true（升序），降序传 false
排序是宽依赖，会触发 Shuffle

3.2 行动算子（Action）

① collect() 方法—— 收集所有数据

作用：将 RDD 中所有元素收集到 Driver 端，返回数组。类型：行动操作（触发计算）

两种用法：

用法1：直接收集（最常用）

scala

rdd.collect()
// 返回 Array[T]

用法2：带偏函数的 collect（了解）

scala

// 定义一个偏函数
val one: PartialFunction[Int, String] = {
  case 1 => "one"
  case _ => "other"
}

// 创建RDD
val rdd = sc.parallelize(List(2, 3, 1))

// 使用带偏函数的collect
rdd.collect(one).collect()
// 结果：Array("other", "other", "one")

⚠️ 重要警告：
collect() 会把所有数据拉到 Driver 端内存
数据量大时会导致 Driver 端内存溢出（OOM）
生产环境大数据量不要用 collect()，用 take() 或直接写入文件
只适合小数据量测试、调试时使用

② take() 方法—— 取前 N 个

作用：获取 RDD 的前 N 个元素，返回数组。类型：行动操作

示例代码：

scala

// 创建RDD（1到10）
val rdd = sc.parallelize(1 to 10)

// 获取前5个元素
rdd.take(5)
// 结果：Array(1, 2, 3, 4, 5)

💡 take vs collect 对比：
方法获取数据量适用场景安全性
collect() 全部小数据量大数据量容易OOM
take(N) 前N个查看样本数据安全，只取少量

方法	获取数据量	适用场景	安全性
collect()	全部	小数据量	大数据量容易OOM
take(N)	前N个	查看样本数据	安全，只取少量

③ 其他常用行动算子

算子	作用	示例
`count()`	返回元素个数	`rdd.count()`
`first()`	返回第一个元素	`rdd.first()`
`reduce(func)`	聚合所有元素	`rdd.reduce(_ + _)`
`foreach(func)`	每个元素执行函数	`rdd.foreach(println)`
`saveAsTextFile(path)`	保存为文本文件	`rdd.saveAsTextFile("/out")`

3.3 转换 vs 行动快速判断

判断方法	转换算子	行动算子
返回值类型	返回 RDD	返回数字/数组/Unit
是否触发计算	否（懒执行）	是（触发Job）
可以链式调用	可以（一直.下去）	调用后就结束了

💡 记忆技巧：看到返回值还是 RDD，那就是转换算子；看到返回的是具体结果或写文件，就是行动算子。

第四部分：RDD 集合操作

RDD 支持一些类似数学集合的操作：并集、交集、补集、笛卡儿积。

方法	数学概念	说明
`union()`	并集	两个RDD合并，不去重
`intersection()`	交集	两个RDD共同的元素
`subtract()`	补集/差集	去掉在另一个RDD中出现的元素
`cartesian()`	笛卡儿积	两两组合，元素数相乘

4.1 union() 方法—— 并集

作用：将两个 RDD 合并成一个，不去重。要求：两个 RDD 的元素类型必须一致。

示例代码：

scala

// 创建两个RDD
val rdd1 = sc.parallelize(List(('a', 1), ('b', 2), ('c', 3)))
val rdd2 = sc.parallelize(List(('a', 1), ('d', 4), ('e', 5)))

// 合并两个RDD
rdd1.union(rdd2).collect()
// 结果：Array(('a',1), ('b',2), ('c',3), ('a',1), ('d',4), ('e',5))
// 注意：('a',1) 出现了两次，union不去重！

⚠️ 注意：union 不去重，如果需要去重，后面加 .distinct()。

4.2 intersection() 方法—— 交集

作用：找出两个 RDD 中共同的元素。特点：会去重，宽依赖（触发 Shuffle）。

示例代码：

scala

// 创建两个RDD
val rdd1 = sc.parallelize(List(('a', 1), ('a', 1), ('b', 1), ('c', 1)))
val rdd2 = sc.parallelize(List(('a', 1), ('b', 1), ('d', 1)))

// 求交集
rdd1.intersection(rdd2).collect()
// 结果：Array(('a', 1), ('b', 1))

4.3 subtract() 方法—— 差集/补集

作用：去掉前一个 RDD 中在后一个 RDD 里出现过的元素。特点：顺序敏感，rdd1.subtract(rdd2) 和 rdd2.subtract(rdd1) 结果不一样。

示例代码：

scala

// 创建两个RDD
val rdd1 = sc.parallelize(List(('a', 1), ('b', 1), ('c', 1)))
val rdd2 = sc.parallelize(List(('c', 1), ('d', 1), ('e', 1)))

// rdd1减去rdd2（rdd1有但rdd2没有的）
rdd1.subtract(rdd2).collect()
// 结果：Array(('a', 1), ('b', 1))

// rdd2减去rdd1（rdd2有但rdd1没有的）
rdd2.subtract(rdd1).collect()
// 结果：Array(('d', 1), ('e', 1))

💡 理解：subtract 就像"减法"，A - B 就是从 A 里去掉 B 有的元素。

4.4 cartesian() 方法—— 笛卡儿积

作用：两个 RDD 的元素两两组合，生成所有可能的配对。特点：结果元素数 = RDD1元素数 × RDD2元素数，数据量会爆炸式增长，慎用！

示例代码：

scala

// 创建两个RDD
val rdd1 = sc.makeRDD(List(1, 3, 5))  // 3个元素
val rdd2 = sc.makeRDD(List(2, 4, 6))  // 3个元素

// 求笛卡儿积
rdd1.cartesian(rdd2).collect()
// 结果：3×3=9个元素
// Array((1,2), (1,4), (1,6), (3,2), (3,4), (3,6), (5,2), (5,4), (5,6))

⚠️ 警告：笛卡儿积数据量是乘积关系，两个1万条的RDD笛卡儿积后就是1亿条！生产环境慎用。

第五部分：键值对 RDD 操作（重点）

5.1 什么是键值对 RDD？

键值对 RDD（Pair RDD）：每个元素都是一个二元组 (Key, Value) 的 RDD。

键（Key）：用来分组、聚合、连接的依据
值（Value）：要处理的数据

为什么需要键值对 RDD？

很多操作都是按 Key 来做的：

按 Key 分组（groupByKey）
按 Key 聚合（reduceByKey）
按 Key 连接（join）

普通 RDD 做不了这些，必须转成键值对 RDD 才行。

5.2 创建键值对 RDD

方式一：用 map() 转换（最常用）

把普通 RDD 的每个元素变成 (Key, Value) 的二元组。

示例代码：

scala

// 创建普通RDD（每行字符串）
val rdd = sc.parallelize(List(
  "this is a test",
  "how are you",
  "I am fine",
  "can you tell me"
))

// 用map创建键值对RDD：第一个单词为Key，整句为Value
val kv_rdd = rdd.map(x => (x.split(" ")(0), x))

// 查看结果
kv_rdd.collect()
// 结果：
// Array(
//   ("this", "this is a test"),
//   ("how", "how are you"),
//   ("I", "I am fine"),
//   ("can", "can you tell me")
// )

方式二：直接创建

创建集合时直接用二元组。

scala

val kv_rdd = sc.parallelize(List(('x', 1), ('x', 2), ('y', 1)))

5.3 keys 和 values 方法

键值对 RDD 可以分别获取所有的键和所有的值。

示例代码：

scala

// 获取所有的键
val key = kv_rdd.keys
key.collect()
// 结果：Array("this", "how", "I", "can")

// 获取所有的值
val value = kv_rdd.values
value.collect()
// 结果：Array("this is a test", "how are you", "I am fine", "can you tell me")

5.4 reduceByKey() 方法—— 按 Key 聚合（重点！）

作用：对相同 Key 的值进行聚合运算。类型：转换操作（宽依赖，触发 Shuffle） 底层原理：先在每个分区内局部聚合（Combiner），再全局聚合，减少数据传输。

示例代码：

scala

// 创建键值对RDD
val rdd = sc.parallelize(List(('x', 1), ('x', 2), ('y', 1), ('z', 1), ('z', 1)))

// 按Key对值进行求和
val reduce_rdd = rdd.reduceByKey((a, b) => a + b)

// 查看结果
reduce_rdd.collect()
// 结果：Array(('x', 3), ('y', 1), ('z', 2))

reduceByKey 的执行过程

以 ('x', 1), ('x', 2) 求和为例：

第1步：取前两个值 a=1, b=2 → 函数 a+b → 结果 A=3
第2步：如果还有下一个值 c，就用 A 和 c 继续算
...
直到每个Key只剩一个值

💡 重点理解：
reduceByKey 是转换算子，不是行动算子！
它会自动做局部聚合（在每个分区内先聚合一次），减少 Shuffle 数据量
这也是 reduceByKey 比 groupByKey 性能好的原因

5.5 groupByKey() 方法—— 按 Key 分组

作用：把相同 Key 的所有值放到一个集合里。类型：转换操作（宽依赖，触发 Shuffle） 返回类型：(K, Iterable[V])

示例代码：

scala

// 还是上面那个rdd
val rdd = sc.parallelize(List(('x', 1), ('x', 2), ('y', 1), ('z', 1), ('z', 1)))

// 按Key分组
val group_rdd = rdd.groupByKey()

// 查看分组结果
group_rdd.collect()
// 结果：Array(('x', CompactBuffer(1, 2)), ('y', CompactBuffer(1)), ('z', CompactBuffer(1, 1)))

// 统计每组值的数量
group_rdd.map(x => (x._1, x._2.size)).collect()
// 结果：Array(('x', 2), ('y', 1), ('z', 2))

reduceByKey vs groupByKey 对比（面试常问！）

对比项	reduceByKey	groupByKey
作用	按Key聚合，直接出结果	按Key分组，值放一起
是否局部聚合	✅ 有（Combiner）	❌ 没有
Shuffle数据量	小（局部聚合后再传）	大（所有数据都要传）
性能	好	差
返回值	(K, V) 聚合后的值	(K, Iterable[V]) 所有值的集合
适用场景	求和、求最值等聚合	需要看到所有值的场景

⚠️ 重要结论：能用 reduceByKey 就不要用 groupByKey！reduceByKey 性能更好。
💡 形象比喻：
reduceByKey：每个车间先统计自己车间的产量，再把各车间的汇总报上去（数据量小）
groupByKey：把所有产品都运到总部再统计（数据量大，网络压力大）

第六部分：RDD 连接操作

连接操作是两个键值对 RDD 按 Key 进行关联，类似 SQL 中的 JOIN。

连接方法	SQL对应	说明
`join()`	INNER JOIN	内连接，只保留两边都有的Key
`leftOuterJoin()`	LEFT JOIN	左外连接，左边RDD的Key全保留
`rightOuterJoin()`	RIGHT JOIN	右外连接，右边RDD的Key全保留
`fullOuterJoin()`	FULL JOIN	全外连接，两边Key都保留

6.1 join() 方法—— 内连接

作用：两个 RDD 中都存在的 Key 才会连接，类似 SQL 的 INNER JOIN。

示例代码：

scala

// 创建两个键值对RDD
val rdd1 = sc.parallelize(List(('a', 1), ('b', 2), ('c', 3)))
val rdd2 = sc.parallelize(List(('a', 1), ('d', 4), ('e', 5)))

// 内连接
val join_rdd = rdd1.join(rdd2)

// 查看结果
join_rdd.collect()
// 结果：Array(('a', (1, 1)))
// 只有Key 'a' 在两边都有，所以只有一条结果
// 值的结构是 (左边的值, 右边的值)

💡 理解：内连接就是"找交集"，两边都有的 Key 才保留。

6.2 leftOuterJoin() 方法—— 左外连接

作用：左边 RDD 的 Key 全部保留，右边 RDD 匹配不上的就是 None。 返回值结构：(Key, (左边值, Option[右边值]))

示例代码：

scala

// 左外连接
val left_join = rdd1.leftOuterJoin(rdd2)

// 查看结果
left_join.collect()
// 结果：
// Array(
//   ('a', (1, Some(1))),   // 匹配上了，右边是Some
//   ('b', (2, None)),      // 没匹配上，右边是None
//   ('c', (3, None))       // 没匹配上，右边是None
// )

💡 Option 类型：
Some(值) 表示有值
None 表示没有值
用 .getOrElse(默认值) 可以安全取值

6.3 rightOuterJoin() 方法—— 右外连接

作用：右边 RDD 的 Key 全部保留，左边匹配不上的就是 None。

示例代码：

scala

// 右外连接
val right_join = rdd1.rightOuterJoin(rdd2)

// 查看结果
right_join.collect()
// 结果：
// Array(
//   ('a', (Some(1), 1)),   // 匹配上了
//   ('d', (None, 4)),      // 左边没有
//   ('e', (None, 5))       // 左边没有
// )

6.4 fullOuterJoin() 方法—— 全外连接

作用：两边 RDD 的 Key 全部保留，哪边匹配不上就是 None。

示例代码：

scala

// 全外连接
val full_join = rdd1.fullOuterJoin(rdd2)

// 查看结果
full_join.collect()
// 结果：
// Array(
//   ('a', (Some(1), Some(1))),  // 两边都有
//   ('b', (Some(2), None)),     // 只有左边有
//   ('c', (Some(3), None)),     // 只有左边有
//   ('d', (None, Some(4))),     // 只有右边有
//   ('e', (None, Some(5)))      // 只有右边有
// )

6.5 四种连接对比总结

连接类型	保留哪些Key	左边没有	右边没有	用途
join（内连接）	两边都有的	-	-	找交集
leftOuterJoin	左边全部	-	None	左表全保留
rightOuterJoin	右边全部	None	-	右表全保留
fullOuterJoin	两边全部	None	None	全保留

💡 记忆技巧：和 SQL 的 JOIN 完全一样，学过 SQL 的话秒懂。

6.6 zip() 方法—— 拉链

作用：把两个 RDD 按位置一一配对，组合成键值对 RDD。要求：两个 RDD 的分区数必须相同，元素个数必须相同，否则报错！

示例代码：

scala

// 创建两个RDD（元素个数和分区数都相同）
val rdd1 = sc.makeRDD(1 to 5, 2)      // 5个元素，2个分区
val rdd2 = sc.makeRDD(Seq("A", "B", "C", "D", "E"), 2)  // 5个元素，2个分区

// zip：rdd1的第n个和rdd2的第n个配对
rdd1.zip(rdd2).collect()
// 结果：Array((1,"A"), (2,"B"), (3,"C"), (4,"D"), (5,"E"))

// 反过来也可以
rdd2.zip(rdd1).collect()
// 结果：Array(("A",1), ("B",2), ("C",3), ("D",4), ("E",5))

⚠️ 注意：zip 要求两个 RDD 分区数和元素数都相同，否则会抛异常。
💡 理解：zip 就像"拉拉链"，左右两边的齿一一对应扣上。

6.7 combineByKey() 方法—— 高级聚合（难点）

什么是 combineByKey？

combineByKey 是 Spark 中最核心的高级聚合方法，很多其他方法（reduceByKey、groupByKey 等）底层都是用它实现的。

特点：可以返回与输入类型不同的结果。

三个核心参数

参数	签名	作用	执行时机
createCombiner	`V => C`	创建累加器初始值	每个分区内，每个Key第一次出现时
mergeValue	`(C, V) => C`	分区内合并：把新值合并到累加器	每个分区内，Key后续出现时
mergeCombiners	`(C, C) => C`	分区间合并：把两个累加器合并	所有分区结果汇总时

💡 通俗理解：
createCombiner：每个分区里，第一次见到这个 Key，先建个"小账本"
mergeValue：同一个分区里再见到这个 Key，把数记到"小账本"上
mergeCombiners：所有分区的"小账本"汇总成"总账本"

示例：用 combineByKey 求平均值

scala

// 创建RDD
val rdd = sc.parallelize(List(
  ("panda", 1), ("panda", 8),
  ("pink", 4), ("pink", 8),
  ("pirate", 5)
))

// 使用combineByKey()方法求每个Key的平均值
val cb_rdd = rdd.combineByKey(
  // 1. createCombiner：第一次见到这个Key，创建初始值(值, 计数=1)
  count => (count, 1),
  
  // 2. mergeValue：分区内合并，累加值，计数+1
  (acc: (Int, Int), count) => (acc._1 + count, acc._2 + 1),
  
  // 3. mergeCombiners：分区间合并，两个累加器相加
  (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
)

// 计算平均值：总和 / 个数
cb_rdd.map(x => (x._1, x._2._1.toDouble / x._2._2)).collect()
// 结果：
// Array(
//   ("panda", 4.5),    // (1+8)/2 = 4.5
//   ("pink", 6.0),     // (4+8)/2 = 6.0
//   ("pirate", 5.0)    // 5/1 = 5.0
// )

💡 理解 combineByKey 的三步：
建账本：每个分区每个 Key 第一次出现时，创建一个累加器
分区内记账：同一个分区内，把新的数据记到账本上
汇总账本：所有分区的账本汇总到一起

6.8 lookup() 方法—— 查找指定 Key 的值

作用：返回键值对 RDD 中指定 Key 的所有对应值。类型：行动操作

示例代码：

scala

// 查询键为"panda"的所有值
rdd.lookup("panda")
// 结果：WrappedArray(1, 8)

第七部分：RDD 文件读写

Spark 支持多种文件格式的读写：

文件格式	读取方法	写入方法	说明
纯文本	`textFile()`	`saveAsTextFile()`	最常用，每行一个元素
JSON	textFile + JSON解析库	转字符串 + saveAsTextFile	半结构化数据
CSV	textFile + CSV解析库	CSVWriter + saveAsTextFile	逗号分隔
SequenceFile	`sequenceFile()`	`saveAsSequenceFile()`	Hadoop键值对格式
ObjectFile	`objectFile[T]()`	`saveAsObjectFile()`	Java序列化对象

7.1 纯文本文件读写

读取文本文件

scala

// 读取HDFS上的文本文件
val rdd = sc.textFile("/tipdm/data/bigdata.txt")

// 查看内容
rdd.collect()

存储文本文件

scala

// 保存为文本文件（repartition(1)表示合并成1个文件）
rdd.repartition(1).saveAsTextFile("/tipdm/data/bigdata_out")

💡 说明：
textFile() 每行是 RDD 的一个元素
saveAsTextFile() 保存的是一个目录，里面有 part-00000、part-00001 等文件
repartition(1) 可以把所有数据合并到一个文件里（小数据量时用）

7.2 JSON 文件读写

什么是 JSON？

JSON（JavaScript Object Notation）是一种轻量级的文本数据交换格式，使用名称/值对的形式。

json

{"name": "Alice", "age": 20}
{"name": "Bob", "age": 21}

读取 JSON 文件

Spark 读取 JSON 的方式是：先当文本文件读，再用 JSON 解析库解析。

要求：文件每行是一条 JSON 记录（不能跨行）。

示例代码：

scala

// 导入JSON4S相关包
import org.json4s._
import org.json4s.jackson.JsonMethods._

// 读取JSON文件（先当文本读）
val data_json = sc.textFile("/tipdm/data/testjson.json")

// 定义样例类（对应JSON结构）
case class Person(name: String, age: Int)

// 定义隐式参数（解析需要）
implicit val formats = DefaultFormats

// 解析JSON
val data = data_json.collect().map { x =>
  parse(x).extract[Person]
}

💡 说明：
用 org.json4s 库解析 JSON
必须定义样例类，JSON 会映射成样例类对象
implicit val formats = DefaultFormats 是必须的隐式参数

存储 JSON 文件

把对象转成 JSON 字符串，再当文本文件保存。

scala

// 导入JSON DSL
import org.json4s.JsonDSL._

// 将对象转成JSON字符串
val data_json = data.map { x =>
  ("name" -> x.name) ~ ("age" -> x.age)
}
val data_string = data_json.map { x =>
  compact(render(x))
}

// 保存为JSON文件
sc.parallelize(data_string.toSeq)
  .repartition(1)
  .saveAsTextFile("/tipdm/data/json_out")

7.3 CSV 文件读写

什么是 CSV？

CSV（Comma Separated Values）：逗号分隔值文件，每行有固定数目的字段，字段间用逗号隔开。

id,name,age
1,Alice,20
2,Bob,21

读取 CSV 文件（简单方式）

如果字段里没有换行符，可以直接用 textFile 读取后 split。

scala

// 读取CSV文件
val data_csv = sc.textFile("/dsj/data/testcsv.csv")

// 简单解析：按逗号分割
val data = data_csv.map(line => line.split(","))
data.collect()

⚠️ 问题：如果字段内容里包含逗号或换行符，简单 split 会出错。

读取 CSV 文件（标准方式：用 CSVReader）

使用 OpenCSV 库解析，能正确处理字段内嵌逗号、引号、换行符。

scala

// 导入相关包
import java.io.StringReader
import au.com.bytecode.opencsv.CSVReader

// 读取CSV文件
val data_csv = sc.textFile("/dsj/data/testcsv.csv")

// 用CSVReader解析
val data = data_csv.map { line =>
  val reader = new CSVReader(new StringReader(line))
  reader.readNext()
}
data.collect()

读取含换行符的 CSV 文件

如果字段里嵌有换行符，需要用 wholeTextFiles 读取整个文件再解析。

scala

import java.io.StringReader
import au.com.bytecode.opencsv.CSVReader
import scala.jdk.CollectionConverters._

// 定义数据结构
case class Data(index: String, title: String, content: String)

// 读取整个文件（wholeTextFiles返回(文件名, 文件内容)）
val data_csv = sc.wholeTextFiles("/dsj/data/testcsv.csv")

// 解析CSV
val data = data_csv.flatMap { case (_, txt) =>
  val reader = new CSVReader(new StringReader(txt))
  reader.readAll().asScala.map(x => Data(x(0), x(1), x(2)))
}
data.collect()

存储 CSV 文件

scala

import java.io.{StringReader, StringWriter}
import au.com.bytecode.opencsv.{CSVReader, CSVWriter}
import scala.jdk.CollectionConverters._

// 将数据转成CSV格式并保存
data.map(data => List(data.index, data.title, data.content).toArray)
  .mapPartitions { data =>
    val stringWriter = new StringWriter()
    val csvWriter = new CSVWriter(stringWriter)
    csvWriter.writeAll(data.toList.asJava)
    Iterator(stringWriter.toString)
  }
  .saveAsTextFile("/dsj/data/csv_out")

7.4 SequenceFile 文件读写

什么是 SequenceFile？

SequenceFile 是 Hadoop 的一种键值对格式的二进制文件，由无固定结构的键值对组成。

存储 SequenceFile

scala

// 导入Hadoop的Writable类型
import org.apache.hadoop.io.{IntWritable, Text}

// 创建键值对RDD
val rdd = sc.parallelize(List(("Panda", 3), ("Monkey", 6), ("Snail", 2)))

// 保存为SequenceFile
rdd.repartition(1).saveAsSequenceFile("/dsj/data/Sequence_out")

💡 说明：
键会自动转成 Text 类型
值会自动转成 IntWritable 类型
文件是二进制格式，直接看是乱码

读取 SequenceFile

scala

// 读取SequenceFile，指定键和值的Writable类型
val data = sc.sequenceFile(
  "/dsj/data/Sequence_out",
  classOf[Text],        // 键的类型
  classOf[IntWritable]  // 值的类型
).map {
  case (x, y) => (x.toString, y.get())  // 转成普通类型
}

// 打印结果
data.collect().foreach(println)

💡 注意：必须指定 keyClass 和 valueClass，而且必须是 Hadoop 的 Writable 类型。

7.5 ObjectFile 文件读写

什么是 ObjectFile？

ObjectFile 是 Spark 特有的格式，基于 Java 序列化，把对象直接存成二进制文件。

对比项	SequenceFile	ObjectFile
格式	Hadoop键值对格式	Java序列化对象
接口要求	实现Writable接口	实现Serializable接口
复杂度	结构复杂	简单直接
适用场景	Hadoop生态通用	Spark内部使用

存储 ObjectFile

scala

// 定义样例类（Scala样例类自动实现Serializable）
case class Person(name: String, age: Int)

// 创建RDD
val rdd = sc.parallelize(Seq(
  Person("Alice", 18),
  Person("Bob", 19),
  Person("Charlie", 20)
))

// 保存为ObjectFile
rdd.repartition(1).saveAsObjectFile("/dsj/data/Object_out")

💡 说明：Scala 的样例类（case class）自动实现了 Serializable 接口，可以直接序列化。

读取 ObjectFile

scala

// 读取ObjectFile，必须指定类型[Person]
val data = sc.objectFile[Person]("/dsj/data/Object_out")

// 打印结果
data.collect().foreach(println)

⚠️ 注意：读取时必须指定泛型类型 [Person]，否则不知道怎么反序列化。

第八部分：项目实战——员工薪资数据查询统计

以下操作在 spark-shell 中执行，数据文件已上传到 HDFS 的 /tipdm/data/ 目录下。

8.0 数据准备

数据文件

Employee_salary_first_half.csv：2023年上半年薪资
Employee_salary_second_half.csv：2023年下半年薪资

字段说明（10个字段，逗号分隔）

索引	字段	含义
0	EmpID	员工编号
1	Name	员工姓名
2	Gender	性别
3	Date_of_Birth	出生日期
4	Age	年龄
5	GROSS	应发工资
6	Net_Pay	实际薪资
7	Deduction	扣款
8	Designation	职位
9	Department	部门

上传数据到 HDFS

先xftp上传到Linux

bash

# 在Linux命令行执行（不是spark-shell里）
hdfs dfs -mkdir -p /dsj/data

hdfs dfs -put /root/data/Employee_salary_first_half.csv /dsj/data/
hdfs dfs -put /root/data/Employee_salary_second_half.csv /dsj/data/

任务3.1：读取员工薪资数据创建 RDD

需求

读取上半年和下半年薪资数据，创建两个 RDD，查看前5行数据。

实现代码

scala

// 读取HDFS数据创建RDD
val first_half = sc.textFile("/dsj/data/Employee_salary_first_half.csv")
val second_half = sc.textFile("/dsj/data/Employee_salary_second_half.csv")

// 查看数据前五行
first_half.take(5).foreach(println)
second_half.take(5).foreach(println)

代码说明

代码	说明
`sc.textFile(path)`	读取HDFS文件，每行一个元素
`.take(5)`	取前5个元素（行动算子）
`.foreach(println)`	每个元素打印一行

任务3.2：查询上半年实际薪资 Top3 的员工

需求

找出上半年实际薪资排名前 3 的员工姓名和薪资。

实现步骤

读取 CSV 文件，删除第一行表头
按逗号分割，取出第2列（姓名）和第7列（实际薪资）
按实际薪资降序排序
取前 3 名

实现代码

scala

// 1. 读取HDFS数据创建RDD
val first_half = sc.textFile("/dsj/data/Employee_salary_first_half.csv")

// 2. 去除首行表头
val header = first_half.first()
val drop_first = first_half.filter(line => line != header)

// 3. 分割，取出姓名（第2列，索引1）和实际薪资（第7列，索引6）
//    注意：薪资转成Int类型才能排序
val split_first = drop_first.map(line => {
  val data = line.split(",")
  (data(1), data(6).toInt)
})

// 4. 按实际薪资降序排序
val sort_first = split_first.sortBy(x => x._2, false)

// 5. 取前3名
sort_first.take(3).foreach(println)

代码说明

代码	说明
`first_half.first()`	取第一行（表头），行动算子
`filter(line => line != header)`	过滤掉表头行
`data(1)`	第2列（姓名），索引从0开始
`data(6).toInt`	第7列（薪资），转成Int才能数值排序
`sortBy(x => x._2, false)`	按第二个值（薪资）降序
`take(3)`	取前3个

⚠️ 常见坑：
忘记去表头，导致表头参与计算（字符串转Int会报错）
薪资没转成 Int，按字符串排序会出问题（"10000" < "9999"）
列索引搞错，第N列的索引是 N-1

任务3.3：查询上半年或下半年薪资大于20万的员工

需求

找出上半年或下半年实际薪资大于 20 万元的员工姓名，去重后输出。

实现步骤

分别读取上下半年数据，去表头，分割
分别过滤出薪资 > 200000 的员工姓名
合并两个 RDD（union）
去重（distinct）

实现代码

scala

// 1. 读取数据
val first_half = sc.textFile("/dsj/data/Employee_salary_first_half.csv")
val second_half = sc.textFile("/dsj/data/Employee_salary_second_half.csv")

// 2. 去表头
val header1 = first_half.first()
val drop_first = first_half.filter(line => line != header1)
val header2 = second_half.first()
val drop_second = second_half.filter(line => line != header2)

// 3. 分割，取姓名和薪资
val split_first = drop_first.map(line => {
  val data = line.split(",")
  (data(1), data(6).toInt)
})
val split_second = drop_second.map(line => {
  val data = line.split(",")
  (data(1), data(6).toInt)
})

// 4. 过滤出薪资>20万的，只保留姓名
val filter_first = split_first.filter(x => x._2 > 200000).map(x => x._1)
val filter_second = split_second.filter(x => x._2 > 200000).map(x => x._1)

// 5. 合并 + 去重
val name = filter_first.union(filter_second).distinct()

// 6. 输出结果
name.collect().foreach(println)

代码说明

代码	说明
`filter(x => x._2 > 200000)`	过滤薪资大于20万的
`.map(x => x._1)`	只保留姓名
`.union(filter_second)`	合并两个RDD（不去重）
`.distinct()`	去重

💡 理解："上半年或下半年" → 只要有一次满足就算 → union 后去重。

任务3.4：统计每位员工 2023 年的总实际薪资

需求

统计每位员工全年（上半年 + 下半年）的总实际薪资。

实现步骤

合并上下半年的薪资数据（union）
按员工姓名分组求和（reduceByKey）

实现代码

scala

// 1. 合并两个RDD（split_first和split_second是任务3.3创建的）
val all_salary = split_first.union(split_second)

// 2. 按姓名（Key）聚合薪资（Value）
val salary = all_salary.reduceByKey((a, b) => a + b)

// 3. 输出结果
salary.collect().foreach(println)

代码说明

代码	说明
`split_first.union(split_second)`	合并上下半年数据，每个员工出现2次
`reduceByKey((a, b) => a + b)`	按姓名分组，把薪资加起来

💡 理解：
union 后每个员工有两条记录（上半年一条、下半年一条）
reduceByKey 按姓名把两条记录的薪资加起来，得到全年总薪资

任务3.5：查询每位员工 2023 年的月均实际薪资

需求

计算每位员工 2023 年的月均实际薪资（总薪资 ÷ 12个月）。

实现步骤

合并上下半年数据
用 combineByKey 计算每个员工的总薪资和记录数
总薪资 ÷ 12 = 月均薪资

实现代码

scala

// 1. 合并两个RDD
val salary = split_first.union(split_second)

// 2. 用combineByKey计算（总薪资，计数）
val cb_salary = salary.combineByKey(
  // 第一次见到这个Key：创建初始值(薪资, 0)
  count => (count, 0),
  // 分区内合并：累加薪资，计数不变（因为每半年一条，所以计数加0）
  (acc: (Int, Int), count) => (acc._1 + count, acc._2 + 0),
  // 分区间合并：两个累加器相加
  (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
)

// 3. 计算月均薪资：总薪资 / 12
val avg_salary = cb_salary.map(x => (x._1, x._2._1.toDouble / 12))

// 4. 输出结果
avg_salary.collect().foreach(println)

代码说明

代码	说明
`combineByKey(...)`	高级聚合，返回(总薪资, 计数)
`x._2._1.toDouble / 12`	总薪资转Double后除以12

💡 说明：
这里 combineByKey 的第二个参数里 acc._2 + 0 看起来奇怪，是因为教材里的写法
实际上每个员工上半年一条、下半年一条，所以每个Key就2条记录
更简单的写法：直接用 reduceByKey 求和后除以 12
教材用 combineByKey 是为了演示这个高级算子的用法

任务3.6：将汇总后的员工薪资存储为文本文件

需求

把员工姓名、上半年薪资、下半年薪资、全年总薪资、月均薪资汇总，保存为文本文件。

实现步骤

准备各个 RDD（上半年薪资、下半年薪资、总薪资、月均薪资）
用 join 把所有数据按姓名连接起来
转换成字符串格式
保存为文本文件

实现代码

scala

// 1. 读取数据、去表头、分割（前面任务已做过）
val first_half = sc.textFile("/tipdm/data/Employee_salary_first_half.csv")
val second_half = sc.textFile("/tipdm/data/Employee_salary_second_half.csv")

val header1 = first_half.first()
val drop_first = first_half.filter(line => line != header1)
val header2 = second_half.first()
val drop_second = second_half.filter(line => line != header2)

val split_first = drop_first.map(line => {
  val data = line.split(",")
  (data(1), data(6).toInt)
})
val split_second = drop_second.map(line => {
  val data = line.split(",")
  (data(1), data(6).toInt)
})

// 2. 计算总薪资
val all_salary = split_first.union(split_second)
val salary = all_salary.reduceByKey((a, b) => a + b)

// 3. 计算月均薪资
val cb_salary = all_salary.combineByKey(
  count => (count, 0),
  (acc: (Int, Int), count) => (acc._1 + count, acc._2 + 0),
  (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
)
val avg_salary = cb_salary.map(x => (x._1, x._2._1.toDouble / 12))

// 4. 多次join，把所有数据连起来
//    split_first.join(split_second) → (姓名, (上半年, 下半年))
//    .join(salary) → (姓名, ((上半年, 下半年), 总薪资))
//    .join(avg_salary) → (姓名, (((上半年, 下半年), 总薪资), 月均))
val result_data = split_first.join(split_second)
  .join(salary)
  .join(avg_salary)
  .map(x => {
    // x._1 = 姓名
    // x._2._1._1._1 = 上半年薪资
    // x._2._1._1._2 = 下半年薪资
    // x._2._1._2 = 总薪资
    // x._2._2 = 月均薪资
    Array(x._1, x._2._1._1._1, x._2._1._1._2, x._2._1._2, x._2._2).mkString(",")
  })

// 5. 保存为文本文件（合并到1个分区）
result_data.repartition(1).saveAsTextFile("/tipdm/data/result_data")

代码说明

多层 join 的嵌套结构

split_first.join(split_second)
    ↓ 结果结构：(姓名, (上半年, 下半年))
.join(salary)
    ↓ 结果结构：(姓名, ((上半年, 下半年), 总薪资))
.join(avg_salary)
    ↓ 结果结构：(姓名, (((上半年, 下半年), 总薪资), 月均薪资))

取值对应关系

表达式	含义
`x._1`	姓名（Key）
`x._2._1._1._1`	上半年薪资
`x._2._1._1._2`	下半年薪资
`x._2._1._2`	全年总薪资
`x._2._2`	月均薪资

💡 技巧：多层 join 后嵌套很深，取值时从外往内一层一层剥，_1 是Key，_2 是Value。

第九部分：常见问题与排错指南

9.1 数据读取类问题

问题1：textFile 读取文件找不到

现象：报错 FileNotFoundException原因：

路径写错了
文件没上传到 HDFS
本地文件没加 file:// 前缀解决：
用 hdfs dfs -ls 路径 确认文件存在
本地文件加 file:// 前缀
注意相对路径和绝对路径的区别

问题2：CSV 数据分割不对

现象：split(",") 后字段数量不对原因：字段内容里包含逗号解决：用 OpenCSV 的 CSVReader 来解析，不要自己 split

问题3：第一行表头参与计算

现象：报 NumberFormatException（字符串转Int失败）原因：忘记去表头解决：

scala

val header = rdd.first()
val data = rdd.filter(line => line != header)

9.2 类型相关问题

问题4：排序结果不对

现象：数字排序变成了字典序（"10000" < "9999"）原因：用字符串排序，不是数值排序解决：排序前转成数值类型

scala

// 错误：字符串排序
rdd.sortBy(x => x._2)

// 正确：转成Int后排序
rdd.map(x => (x._1, x._2.toInt)).sortBy(x => x._2)

问题5：类型不匹配报错

现象：报错 type mismatch原因：RDD 元素类型和期望的不一样解决：

用 rdd.getClass 或打印看看数据类型
注意 String 和 Int/Double 的转换

9.3 连接操作类问题

问题6：join 后数据量比预期少

原因：用的是内连接（join），两边都有的 Key 才保留解决：根据需求选择合适的连接类型

左边全保留 → leftOuterJoin
右边全保留 → rightOuterJoin
两边全保留 → fullOuterJoin

问题7：zip 操作报错

现象：报 Can only zip RDDs with same number of elements in each partition原因：两个 RDD 的分区数或元素数不一样解决：

确保两个 RDD 元素数量相同
确保分区数相同（可以用 repartition() 调整）

9.4 内存相关问题

问题8：collect() 后 Driver 端 OOM

现象：报 OutOfMemoryError原因：数据量太大，全部 collect 到 Driver 端内存不够解决：

大数据量不要用 collect()
用 take() 看少量样本
直接用 saveAsTextFile() 保存到文件

问题9：Executor 内存不足

现象：Executor 挂掉，报 GC overhead 或 OOM 原因：分配的 Executor 内存不够解决：

调大 --executor-memory
增加分区数，每个分区数据量小一些
优化代码，避免数据倾斜

9.5 排错通用思路

看报错信息：先看异常类型和错误消息
看数据：用 take() 看看数据长什么样，是不是和预期一致
看类型：打印一下 RDD 的元素类型，确认类型对不对
分步调试：每一步都看看结果，不要一口气写到底
小数据测试：先用小数据验证逻辑，再跑大数据

💡 排错口诀：「先看报错，再看数据，类型确认，分步调试」

第十部分：实习 / 面试高频考点

10.1 概念类（高频）

Q1：创建 RDD 有哪几种方式？

三种： ① 从内存集合创建（parallelize、makeRDD） ② 从外部存储读取（textFile 等） ③ 从已有 RDD 转换得到（各种转换算子）

Q2：转换算子和行动算子的区别？各举几个例子

转换算子：懒执行，返回新 RDD，如 map、filter、flatMap、reduceByKey、join 行动算子：触发计算，返回结果或写入存储，如 collect、count、take、saveAsTextFile、foreach 判断技巧：返回值是 RDD → 转换；返回值是具体结果 → 行动

Q3：什么是键值对 RDD？为什么需要它？

每个元素都是 (Key, Value) 二元组的 RDD。需要它是因为很多操作（分组、聚合、连接）都是按 Key 来做的，普通 RDD 做不了。

Q4：reduceByKey 和 groupByKey 的区别？哪个性能好？

reduceByKey：按 Key 聚合，有局部聚合（Combiner），Shuffle 数据量小，性能好 groupByKey：按 Key 分组，没有局部聚合，Shuffle 数据量大，性能差结论：能用 reduceByKey 就不要用 groupByKey

Q5：map 和 flatMap 的区别？

map：一对一，输入 N 个输出 N 个，每个元素变成一个结果 flatMap：一对多，先 map 再 flatten，每个元素变成多个结果，最后摊平典型应用：flatMap 用来切单词

Q6：Spark 支持哪些文件格式？

纯文本（textFile）、JSON、CSV、SequenceFile、ObjectFile 等

10.2 操作类（高频）

Q7：怎么找出 RDD 中排名前 N 的元素？

用 sortBy 排序（降序），然后 take(N)
scala
rdd.sortBy(x => x._2, false).take(N)

Q8：怎么对两个 RDD 进行左外连接？

用 leftOuterJoin() 方法
scala
rdd1.leftOuterJoin(rdd2)
结果中右边匹配不上的是 None。

Q9：combineByKey 的三个参数分别是什么作用？

① createCombiner：创建累加器初始值（每个分区每个Key第一次出现时） ② mergeValue：分区内合并（把新值合并到累加器） ③ mergeCombiners：分区间合并（两个累加器合并）

Q10：join 和 union 的区别？

union：合并两个 RDD 的所有元素，不去重，要求两个 RDD 结构一样 join：按键连接两个键值对 RDD，只保留匹配的 Key，结果是 (Key, (左边值, 右边值))

10.3 原理类（中频）

Q11：为什么 reduceByKey 比 groupByKey 性能好？

因为 reduceByKey 有局部聚合（Combiner）：
在每个分区内先做一次聚合，减少了 Shuffle 的数据量
groupByKey 没有局部聚合，所有数据都要通过网络传输
Shuffle 是 Spark 性能瓶颈，减少 Shuffle 数据量就能提升性能

Q12：什么是 Shuffle？哪些算子会触发 Shuffle？

Shuffle 就是数据重新分区（洗牌），需要跨节点传输数据，是性能瓶颈。会触发 Shuffle 的算子：
重分区类：repartition、coalesce（shuffle=true时）
聚合类：reduceByKey、groupByKey、combineByKey
排序类：sortByKey、sortBy
集合类：intersection、subtract、distinct
连接类：join、cogroup

Q13：distinct 是怎么实现的？

distinct 底层是用 map + reduceByKey 实现的：
每个元素变成 (元素, null)
reduceByKey 聚合（相同 Key 只留一个）
再 map 取出 Key 所以 distinct 是宽依赖，会触发 Shuffle。

10.4 实操类（高频）

Q14：读取 CSV 文件需要注意什么？

① 第一行是表头，需要去掉 ② 数值类型的字段要转成 Int/Double 才能计算 ③ 字段里可能有逗号，简单 split 会出错，要用 CSVReader ④ 列索引从 0 开始，第 N 列索引是 N-1

Q15：怎么把 RDD 保存成一个文件？

默认 saveAsTextFile 会保存成多个文件（每个分区一个）。要保存成一个文件，用 repartition(1) 合并成一个分区：
scala
rdd.repartition(1).saveAsTextFile("/path")
注意：只适合小数据量，大数据量不要这么做。

Q16：ObjectFile 和 SequenceFile 的区别？

ObjectFile：Java 序列化对象，简单直接，Spark 特有，要求对象实现 Serializable SequenceFile：Hadoop 键值对格式，Hadoop 生态通用，要求实现 Writable 接口

附录：习题解析

选择题解析

1、答案：C 解析：从外部存储系统中读取数据创建 RDD 可通过 SparkContext 对象的 textFile() 方法读取数据集。parallelize 和 makeRDD 是从内存集合创建的方法。

2、答案：C 解析：Spark RDD 常用的转换操作有 map()、filter()、flatMap()、union()、groupByKey()、reduceByKey()、distinct() 等；常用的行动操作有 reduce()、collect()、count()、first()、take()、saveAsTextFile()、foreach() 等。collect() 是行动操作，不是转换操作。

3、答案：A 解析：map() 是转换操作（懒执行，返回新 RDD）；collect()、count()、take() 都是行动操作（触发计算）。

4、答案：B 解析：sortBy() 方法的第 2 个参数是 ascending，决定排序后 RDD 中的元素是升序还是降序，默认为 true（升序），若需降序则设为 false。

5、答案：C 解析：subtract() 方法用于将原 RDD 里和参数 RDD 里相同的元素去掉（求差集）。union 是并集，intersection 是交集，cartesian 是笛卡儿积。

6、答案：D 解析：join() 方法用于对两个 RDD 进行内连接；leftOuterJoin() 是左外连接；rightOuterJoin() 是右外连接；fullOuterJoin() 是全外连接。四种连接方法都存在。

7、答案：B 解析：reduceByKey(_ + _) 可以将相同键的所有值加起来，此处键 a 对应的值有 1, 2, 3，累加结果为 6，所以最终结果是 (a, 6)。

8、答案：A 解析：join() 方法是内连接，只有当两个 RDD 中存在相同的键时才会产生结果。对于键 a，RDD1 中的 (a, 1) 与 RDD2 中的 (a, 6) 连接，生成 (a, (1, 6))；对于键 b，RDD1 中的 (b, 2) 与 RDD2 中的 (b, 4) 和 (b, 5) 分别连接，生成 (b, (2, 4)) 和 (b, (2, 5))；对于键 c，由于 RDD2 中没有键为 c 的元素，因此不产生结果。

9、答案：D 解析：zip() 方法用于将两个 RDD 组合成键值对 RDD，要求两个 RDD 的分区数和元素数都相同。该方法会将第一个 RDD 的第 n 个元素与第二个 RDD 的第 n 个元素配对，形成新的二元组。

10、答案：D 解析：RDD 数据可以直接调用 saveAsTextFile() 方法将数据存储为文本文件。textFile 是读取方法，不是写入方法。

操作题解析

题目

查找 examination2019.csv 文件中各地区本科批次的分数线，并将结果以文本格式存储到 HDFS。

数据说明

examination2019.csv：2019年各地区高考分数线数据，包含地区、年份、批次、分数等字段。

实现代码

scala

// 1. 上传数据至HDFS（Linux命令行执行）
hdfs dfs -put /opt/data/examination2019.csv /tipdm/data

// 2. spark-shell读取文件
val input = sc.textFile("/tipdm/data/examination2019.csv")

// 3. 查找出各地区本科批次的分数线
//    分割后取第1列(地区)、第2列(年份)、第3列(批次)、第4列(分数)
//    过滤出批次包含"本科批"的记录
val score = input.map(x => {
  val line = x.split(",")
  (line(0), line(1), line(2), line(3))
}).filter(_._3.contains("本科批"))

// 4. 将结果以文本格式存储到HDFS上
score.repartition(1).saveAsTextFile("/tipdm/data/result/examination2019")

代码说明

代码	说明
`x.split(",")`	按逗号分割每行
`(line(0), line(1), line(2), line(3))`	取前4列：地区、年份、批次、分数
`filter(_._3.contains("本科批"))`	过滤批次包含"本科批"的记录
`repartition(1)`	合并成一个分区，输出一个文件
`saveAsTextFile`	保存为文本文件

笔记版本：V1.0 对应教材：《Spark大数据技术与应用（第3版）》人民邮电出版社 对应项目：项目3 查询和统计员工薪资数据——Spark Shell编程 最后更新：2026年8月

项目3：查询和统计员工薪资数据——Spark Shell编程 ​

目录 ​

第一部分：项目背景与 Spark Shell 入门 ​

1.1 项目背景 ​

为什么要学 Spark Shell 编程？ ​

项目数据说明 ​

1.2 Spark Shell 是什么？ ​

特点 ​

启动方式 ​

第二部分：创建 RDD 的三种方式 ​

2.1 方式一：从内存集合创建 RDD ​

① parallelize() 方法 ​

② makeRDD() 方法 ​

parallelize vs makeRDD 对比 ​

2.2 方式二：从外部存储读取创建 RDD ​

支持的数据源 ​

① 读取本地文件 ​

② 读取 HDFS 文件（最常用） ​

2.3 方式三：从已有 RDD 转换得到新 RDD ​

第三部分：RDD 基础操作（转换 + 行动） ​

3.1 转换算子（Transformation） ​

① map() 方法—— 一一映射 ​

② flatMap() 方法—— 先映射再扁平化 ​

③ filter() 方法—— 过滤 ​

④ distinct() 方法—— 去重 ​

⑤ sortBy() 方法—— 排序 ​

3.2 行动算子（Action） ​

① collect() 方法—— 收集所有数据 ​

② take() 方法—— 取前 N 个 ​

③ 其他常用行动算子 ​

3.3 转换 vs 行动 快速判断 ​

第四部分：RDD 集合操作 ​

4.1 union() 方法—— 并集 ​

4.2 intersection() 方法—— 交集 ​

4.3 subtract() 方法—— 差集/补集 ​

4.4 cartesian() 方法—— 笛卡儿积 ​

第五部分：键值对 RDD 操作（重点） ​

5.1 什么是键值对 RDD？ ​

为什么需要键值对 RDD？ ​

5.2 创建键值对 RDD ​

方式一：用 map() 转换（最常用） ​

方式二：直接创建 ​

5.3 keys 和 values 方法 ​

5.4 reduceByKey() 方法—— 按 Key 聚合（重点！） ​

reduceByKey 的执行过程 ​

5.5 groupByKey() 方法—— 按 Key 分组 ​

reduceByKey vs groupByKey 对比（面试常问！） ​

第六部分：RDD 连接操作 ​

6.1 join() 方法—— 内连接 ​

6.2 leftOuterJoin() 方法—— 左外连接 ​

6.3 rightOuterJoin() 方法—— 右外连接 ​

6.4 fullOuterJoin() 方法—— 全外连接 ​

6.5 四种连接对比总结 ​

6.6 zip() 方法—— 拉链 ​

6.7 combineByKey() 方法—— 高级聚合（难点） ​

什么是 combineByKey？ ​

三个核心参数 ​

示例：用 combineByKey 求平均值 ​

6.8 lookup() 方法—— 查找指定 Key 的值 ​

第七部分：RDD 文件读写 ​

7.1 纯文本文件读写 ​

读取文本文件 ​

存储文本文件 ​

7.2 JSON 文件读写 ​

什么是 JSON？ ​

读取 JSON 文件 ​

存储 JSON 文件 ​

7.3 CSV 文件读写 ​

什么是 CSV？ ​

读取 CSV 文件（简单方式） ​

读取 CSV 文件（标准方式：用 CSVReader） ​

读取含换行符的 CSV 文件 ​

存储 CSV 文件 ​

7.4 SequenceFile 文件读写 ​

什么是 SequenceFile？ ​

存储 SequenceFile ​

读取 SequenceFile ​

7.5 ObjectFile 文件读写 ​

什么是 ObjectFile？ ​

存储 ObjectFile ​

项目3：查询和统计员工薪资数据——Spark Shell编程

目录

第一部分：项目背景与 Spark Shell 入门

1.1 项目背景

为什么要学 Spark Shell 编程？

项目数据说明

1.2 Spark Shell 是什么？

特点

启动方式

第二部分：创建 RDD 的三种方式

2.1 方式一：从内存集合创建 RDD

① parallelize() 方法

② makeRDD() 方法

parallelize vs makeRDD 对比

2.2 方式二：从外部存储读取创建 RDD

支持的数据源

① 读取本地文件

② 读取 HDFS 文件（最常用）

2.3 方式三：从已有 RDD 转换得到新 RDD

第三部分：RDD 基础操作（转换 + 行动）

3.1 转换算子（Transformation）

① map() 方法—— 一一映射

② flatMap() 方法—— 先映射再扁平化

③ filter() 方法—— 过滤

④ distinct() 方法—— 去重

⑤ sortBy() 方法—— 排序

3.2 行动算子（Action）

① collect() 方法—— 收集所有数据

② take() 方法—— 取前 N 个

③ 其他常用行动算子

3.3 转换 vs 行动快速判断

第四部分：RDD 集合操作

4.1 union() 方法—— 并集

4.2 intersection() 方法—— 交集

4.3 subtract() 方法—— 差集/补集

4.4 cartesian() 方法—— 笛卡儿积

第五部分：键值对 RDD 操作（重点）

5.1 什么是键值对 RDD？

为什么需要键值对 RDD？

5.2 创建键值对 RDD

方式一：用 map() 转换（最常用）

方式二：直接创建

5.3 keys 和 values 方法

5.4 reduceByKey() 方法—— 按 Key 聚合（重点！）

reduceByKey 的执行过程

5.5 groupByKey() 方法—— 按 Key 分组

reduceByKey vs groupByKey 对比（面试常问！）

第六部分：RDD 连接操作

6.1 join() 方法—— 内连接

6.2 leftOuterJoin() 方法—— 左外连接

6.3 rightOuterJoin() 方法—— 右外连接

6.4 fullOuterJoin() 方法—— 全外连接

6.5 四种连接对比总结

6.6 zip() 方法—— 拉链

6.7 combineByKey() 方法—— 高级聚合（难点）

什么是 combineByKey？

三个核心参数

示例：用 combineByKey 求平均值

6.8 lookup() 方法—— 查找指定 Key 的值

第七部分：RDD 文件读写

7.1 纯文本文件读写

读取文本文件

存储文本文件

7.2 JSON 文件读写

什么是 JSON？

读取 JSON 文件

存储 JSON 文件

7.3 CSV 文件读写

什么是 CSV？

读取 CSV 文件（简单方式）

读取 CSV 文件（标准方式：用 CSVReader）

读取含换行符的 CSV 文件

存储 CSV 文件

7.4 SequenceFile 文件读写

什么是 SequenceFile？

存储 SequenceFile

读取 SequenceFile

7.5 ObjectFile 文件读写

什么是 ObjectFile？

存储 ObjectFile