当前位置：首页 > news >正文

Spark面试题清单

news 2025/9/18 16:15:36

当然可以！以下是一份常见的 Apache Spark 面试题清单（含参考答案），覆盖基础概念、核心原理、性能优化、故障排查、SQL 与 DataFrame、RDD 与 Dataset 等多个维度，适用于初中高级 Spark 开发/数据工程师岗位面试。

Apache Spark 常见面试题 + 参考答案

一、基础概念类

Q1：Spark 是什么？它和 MapReduce 的主要区别是什么？

✅ 答案：

Spark 是一个基于内存计算的分布式计算框架，支持批处理、流处理、机器学习和图计算。
与 MapReduce 区别：

维度	Spark	MapReduce
计算模型	DAG 执行引擎，支持内存迭代	两阶段模型（Map → Reduce），磁盘落地
性能	快（尤其迭代算法）	慢（频繁读写磁盘）
API 丰富度	高（RDD/DataFrame/Dataset/SQL/Streaming/MLlib）	低（仅 Map/Reduce）
容错机制	Lineage（血统）重算	数据落盘恢复
实时性	支持微批流（Structured Streaming）	仅批处理

Spark 不是完全取代 MR，而是更适合需要迭代、交互式分析、实时流的场景。

Q2：Spark 的核心组件有哪些？

✅ 答案：

Spark Core：任务调度、内存管理、容错、存储系统交互等基础功能。
Spark SQL：结构化数据处理，支持 SQL 查询和 DataFrame/Dataset API。
Spark Streaming / Structured Streaming：流式数据处理。
MLlib：机器学习库。
GraphX：图计算库。
Cluster Managers：支持 Standalone、YARN、Mesos、Kubernetes。

二、RDD 相关

Q3：什么是 RDD？它的五大特性是什么？

✅ 答案：

RDD（Resilient Distributed Dataset） 是 Spark 最基本的数据抽象，代表一个不可变、可分区、可并行操作的元素集合。

五大特性（论文中提出）：

一组分区（Partitions）
每个分区上的计算函数（Compute function per partition）
依赖关系（Dependencies on other RDDs）→ Lineage
分区器（Partitioner，可选，用于 KV RDD）
首选位置（Preferred Locations，如 HDFS 块位置）→ 数据本地性

Q4：Transformation 和 Action 的区别？举几个例子。

✅ 答案：

Transformation（转换）：懒执行，返回新 RDD。如：
- map, filter, flatMap, groupByKey, reduceByKey, join, distinct
Action（行动）：触发实际计算，返回结果或写入外部存储。如：
- collect, count, take, saveAsTextFile, foreach

⚠️ Transformation 是 Lazy 的，只有遇到 Action 才会真正执行 Job。

三、Shuffle & 性能优化

Shuffle 是 Spark 中为了实现宽依赖操作（如 groupByKey、join）而进行的数据重分区与跨节点传输过程。它包含 Shuffle Write 和 Shuffle Read 两个阶段，默认会落盘并走网络，带来磁盘 IO、网络开销和序列化成本，是 Spark 最大的性能瓶颈。

常见触发 Shuffle 的操作包括：groupByKey、reduceByKey、join（非广播）、distinct、repartition 等。其中，groupByKey 性能最差，因为它不做预聚合；而 reduceByKey 会在 Map 端先局部聚合，显著减少 Shuffle 数据量。

我们在项目中会尽量避免 Shuffle —— 比如用 reduceByKey 替代 groupByKey、小表用 broadcast join、开启 AQE 自动合并小分区等。”

Q5：什么是 Shuffle？哪些操作会触发 Shuffle？

✅ 答案：

Shuffle 是指数据在不同分区之间重新分布的过程，通常发生在宽依赖（wide dependency）操作中。

常见触发 Shuffle 的操作：

groupByKey
reduceByKey（但可预聚合减少数据量）
join（非广播情况下）
distinct
repartition
sortByKey

Shuffle 是性能瓶颈，应尽量避免或优化。

Q6：如何优化 Spark Shuffle 性能？

✅ 答案：

**使用 reduceByKey / aggregateByKey 替代 **groupByKey → 预聚合减少数据量。
调整 spark.sql.shuffle.partitions（默认200） → 根据数据量合理设置分区数。
开启 Tungsten 引擎和 Kryo 序列化：

spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

增大 Shuffle 内存比例：

spark.conf.set("spark.shuffle.memoryFraction", "0.4") // Spark < 3.0
// Spark 3.0+ 使用统一内存管理，无需手动调此参数

使用广播 Join 替代 Shuffle Join（小表广播）。
启用 AQE（Adaptive Query Execution）（Spark 3.0+）自动优化 Shuffle 分区。

✅ “用广播 Join 替代 Shuffle Join（小表广播）” 是 Spark 性能优化中最常用、最有效的手段之一。

一、什么是 Broadcast Join？

Broadcast Join（广播连接）：将小表的数据广播到所有 Executor 节点的内存中，然后在每个节点上与大表进行本地 Join，完全避免了 Shuffle 操作。

也叫 Map-side Join 或 Hash Join（本地哈希连接）
是 Spark SQL / DataFrame 中的自动优化策略之一（需满足条件）

⚖️ 二、为什么用 Broadcast Join 替代 Shuffle Join？

❗ Shuffle Join 的问题：

largeDF.join(smallDF, "key") // 默认可能走 SortMergeJoin → 触发 Shuffle！

Shuffle 开销巨大：网络传输 + 磁盘IO + 序列化
资源竞争：大量 Task、内存压力、GC 频繁
性能瓶颈：尤其在数据倾斜时，个别 Task 拖慢整个 Job

✅ Broadcast Join 的优势：

largeDF.join(broadcast(smallDF), "key") // 显式广播 → 无 Shuffle！

零 Shuffle → 极大提升性能
本地 Join → 每个分区独立计算，无等待
适合星型模型 → 事实表（大）JOIN 维度表（小）

性能提升可达数倍甚至数十倍！

三、什么时候适合使用 Broadcast Join？

✅ 适用条件：

条件	说明
小表足够小	默认阈值：10MB（可通过 `spark.sql.autoBroadcastJoinThreshold` 设置）
内存充足	Executor 有足够堆外/堆内内存缓存广播变量
Join 类型支持	支持 Inner Join、Left Outer Join、Right Outer Join、Left Semi Join 等

注意：全外连接（Full Outer Join）不支持广播！

️ 四、如何实现 Broadcast Join？

方法1：Spark SQL 自动广播（推荐）

// 设置广播阈值（默认10MB = 10 * 1024 * 1024）
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10485760) // 10MB// 当小表统计信息 < 阈值，Catalyst 优化器会自动选择 BroadcastHashJoin
val result = largeDF.join(smallDF, "id")

→ 查看执行计划确认是否生效：

result.explain()
// 如果看到 BroadcastHashJoin，则成功！

方法2：显式调用 `broadcast()` 函数（强制广播）

import org.apache.spark.sql.functions.broadcastval result = largeDF.join(broadcast(smallDF), "id")

→ 即使小表 > 阈值，也会强制广播！慎用，可能OOM！

五、执行计划对比

➤ Shuffle Join（SortMergeJoin）

== Physical Plan ==
*(3) SortMergeJoin [id#10L], [id#20L], Inner
:- *(1) Sort [id#10L ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(id#10L, 200)
:     +- Scan ExistingRDD...
+- *(2) Sort [id#20L ASC NULLS FIRST], false, 0+- Exchange hashpartitioning(id#20L, 200)+- Scan ExistingRDD...

→ 有 Exchange → 表示发生了 Shuffle！

➤ Broadcast Join（BroadcastHashJoin）

== Physical Plan ==
*(2) BroadcastHashJoin [id#10L], [id#20L], Inner, BuildRight
:- Scan ExistingRDD...  // 大表正常扫描
+- BroadcastExchange HashedRelationBroadcastMode...+- Scan ExistingRDD...  // 小表被广播

→ 有 BroadcastExchange → 无 Shuffle，小表被广播！

⚠️ 六、注意事项 & 最佳实践

1. ❗ 不要广播大表 → 会导致 Executor OOM！

广播变量存储在 Executor 的堆外内存（默认）或堆内存。
若广播 1GB 表到 100 个 Executor → 需要 100GB 内存！

✅ 解决方案：

调整阈值前评估小表大小
使用 df.cache().count() 预先触发统计信息收集
监控 Executor 内存使用（Spark UI → Executors 页面）

2. ✅ 提前收集统计信息（CBO）

smallDF.cache().count() // 触发计算并缓存，同时收集 stats
largeDF.join(smallDF, "id") // Catalyst 更容易选对 Join 策略

→ 否则可能因统计信息缺失，无法自动广播。

3. 动态调整广播阈值

// 临时提高阈值（如50MB），适用于已知小表<50M的情况
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 52428800) // 50MB// 用完后建议恢复默认值，避免误广播
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10485760)

4. 广播变量监控

在 Spark UI → “SQL” Tab → 查看 “BroadcastHashJoin”
在 “Executors” Tab → 查看 “Storage Memory” 使用情况
广播变量会显示为 broadcast_XX

七、实战案例：用户订单关联用户信息

val ordersDF = spark.read.parquet("hdfs://.../orders")    // 大表：10亿行
val usersDF = spark.read.parquet("hdfs://.../users")      // 小表：10万行，约5MB// 方式1：自动广播（推荐）
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10485760)
val result = ordersDF.join(usersDF, "user_id")// 方式2：显式广播（确保生效）
import org.apache.spark.sql.functions.broadcast
val result = ordersDF.join(broadcast(usersDF), "user_id")result.explain() // 确认出现 BroadcastHashJoin

→ 性能从分钟级 → 秒级！

✅ 八、总结：Broadcast Join 使用 Checklist

检查项	是否满足
小表大小 < 广播阈值（默认10MB）？	✅
Join 类型支持广播（非 Full Outer）？	✅
Executor 内存足以缓存广播变量？	✅
已收集小表统计信息？	✅
执行计划确认是 BroadcastHashJoin？	✅
避免在循环/UDF中重复广播？	✅

Bonus：广播变量不止用于 Join！

你还可以手动创建广播变量用于其他场景：

val smallMap = sc.broadcast(Map(1 -> "A", 2 -> "B"))rdd.map { x =>val label = smallMap.value.getOrElse(x.id, "Unknown")(x, label)
}

→ 避免每个 Task 序列化发送 Map，节省网络和序列化开销！

✅ 记住口诀：

“小表广播，大表不动；避免 Shuffle，性能翻倍！”

合理使用 Broadcast Join，是 Spark 调优中最立竿见影的手段之一！

Q7：Spark 如何避免数据倾斜（Data Skew）？

✅ 答案：

数据倾斜表现：少数 Task 处理大量数据，拖慢整个 Job。

解决方案：

加盐（Salting）打散热点 Key：

// 原始 key → 加随机前缀分散到不同分区
val salted = rdd.map { case (key, value) =>(s"${Random.nextInt(10)}_$key", value)
}

两阶段聚合（局部聚合 + 全局聚合）。
采样分析倾斜 Key，单独处理。
使用 repartition 或 coalesce 调整分区策略。
Broadcast Join 小表。
AQE 自动检测并拆分倾斜分区（Spark 3.0+）。

四、内存 & 资源管理

Q8：Spark 的内存模型是怎样的？（Spark 1.x vs 3.x）

✅ 答案：

➤ Spark 1.x：静态内存管理

分为 Storage（缓存）、Execution（Shuffle/聚合）、Other（用户代码）三块，比例固定。

➤ Spark 3.x：统一内存管理（Unified Memory Manager）

Storage + Execution 共享同一块内存池，可动态借用。
默认比例：
- spark.memory.fraction = 0.6 → 总堆内存的60%用于 Storage + Execution
- spark.memory.storageFraction = 0.5 → 其中50%优先用于 Storage

✅ 更灵活，减少 OOM，提升资源利用率。

Q9：Spark 中的 Cache / Persist 有什么作用？级别有哪些？

✅ 答案：

作用：将 RDD/DataFrame 缓存到内存或磁盘，避免重复计算，加速迭代。
调用方式：

rdd.cache()           // = persist(StorageLevel.MEMORY_ONLY)
df.persist(StorageLevel.MEMORY_AND_DISK)

常用 StorageLevel：

级别	说明
`MEMORY_ONLY`	仅内存，无序列化（最快）
`MEMORY_ONLY_SER`	内存，序列化（节省空间）
`MEMORY_AND_DISK`	内存不够时溢写磁盘
`DISK_ONLY`	仅磁盘
`OFF_HEAP`	堆外内存（需配置）

⚠️ 缓存虽好，但占用内存，不用时记得 unpersist() 释放！

五、Spark SQL & DataFrame

Q10：DataFrame 和 RDD 有什么区别？

✅ 答案：

特性	RDD	DataFrame
类型安全	编译期不检查类型（运行时报错）	Schema 结构化，编译期部分检查
优化器	无	Catalyst 优化器（谓词下推、列裁剪等）
执行引擎	通用计算	Tungsten 引擎（高效内存布局 + CodeGen）
API 易用性	函数式，较底层	SQL/DSL，更易用
性能	一般	更高（尤其结构化数据）

推荐：优先使用 DataFrame/Dataset，除非需要高度自定义逻辑。

Q11：Catalyst 优化器的作用是什么？

✅ 答案：

Catalyst 是 Spark SQL 的查询优化器，负责：

解析（Parsing）：SQL → Unresolved Logical Plan
分析（Analysis）：绑定表/列元数据 → Resolved Logical Plan
逻辑优化（Logical Optimization）：如谓词下推、常量折叠、列裁剪
物理计划生成（Physical Planning）：生成多个执行计划，基于 Cost Model 选择最优
代码生成（Whole-stage Code Generation）：将多步操作编译成 Java 字节码，消除虚函数调用开销

→ 大幅提升 SQL 执行效率！

六、容错 & 部署

Q12：Spark 如何实现容错？

✅ 答案：

Lineage（血统）机制：记录 RDD 的转换过程（DAG），当某个分区丢失，可通过父分区重新计算。
Checkpoint（检查点）：对代价高的 RDD（如迭代算法）持久化到可靠存储（HDFS），截断 Lineage。
Executor 失败：Driver 重新调度 Task 到其他 Executor。
Driver 失败：需外部系统（如 YARN/K8s）重启 Driver（Spark 本身不保证 Driver 高可用）。

Q13：Spark on YARN 的 client 模式和 cluster 模式有什么区别？

✅ 答案：

特性	Client 模式	Cluster 模式
Driver 运行位置	提交作业的客户端机器	ApplicationMaster 容器内（集群中）
客户端是否必须在线	是（Driver 在客户端）	否（提交后即可断开）
适用场景	交互式调试、开发	生产环境、后台作业
日志查看	客户端控制台	YARN Web UI / logs

✅ 生产环境推荐 cluster 模式！

七、Structured Streaming

Q14：Structured Streaming 的三种输出模式是什么？

✅ 答案：

Append Mode（追加模式）
→ 只输出新增行（适用于无聚合的流）。
Update Mode（更新模式）
→ 输出有更新的行（适用于聚合流，如 count、sum）。
Complete Mode（完整模式）
→ 每次输出全量结果（适用于聚合且结果集较小的场景）。

⚠️ 并非所有 sink 都支持所有模式（如 Kafka 不支持 Complete Mode）。

Q15：Watermark 的作用是什么？

✅ 答案：

Watermark 用于处理延迟数据和清理状态。

定义“事件时间”的容忍延迟（如 watermark = eventTime - 10 minutes）。
超过 watermark 的数据被视为迟到数据，可选择丢弃或侧输出。
同时，Spark 会自动清理 watermark 之前的聚合状态，防止状态无限增长。

df.withWatermark("eventTime", "10 minutes").groupBy(window($"eventTime", "5 minutes"), $"category").count()

八、综合 & 场景题

Q16：如何监控和调优一个慢的 Spark 作业？

✅ 答案：

查看 Spark UI：
- Stages 页面：看 Task 分布、GC 时间、Shuffle Read/Write
- Executors 页面：看内存/CPU 使用率、数据本地性
检查数据倾斜 → 看个别 Task 运行时间远长于其他。
调整分区数 → spark.sql.shuffle.partitions、repartition()。
优化 Shuffle → 用 reduceByKey、广播 Join。
增加资源 → Executor 内存、Core 数、并行度。
开启 AQE（Spark 3.0+） → 自动优化分区、Join 策略、倾斜处理。
序列化优化 → 使用 Kryo。
缓存中间结果 → 避免重复计算。

Q17：Spark 如何读写 HBase / Kafka / MySQL？

✅ 答案：

HBase：

spark.read.format("org.apache.hadoop.hbase.spark").option("hbase.table", "table").load()

或使用 newAPIHadoopRDD + TableInputFormat

Kafka（Structured Streaming）：

spark.readStream.format("kafka").option("kafka.bootstrap.servers", "...").option("subscribe", "topic").load()

MySQL：

spark.read.format("jdbc").option("url", "jdbc:mysql://...").option("dbtable", "table").option("user", "user").option("password", "pwd").load()

注意：生产环境连接数据库要考虑分区字段、批量读写、连接池等。

✅ 附：高频考点总结

类别	高频考点
基础	Spark vs MR、组件、运行模式
RDD	Transformation vs Action、持久化、依赖关系
Shuffle	触发操作、优化方法、数据倾斜解决
SQL	Catalyst、DataFrame vs RDD、谓词下推
内存	统一内存模型、StorageLevel、OOM 排查
Streaming	Watermark、输出模式、端到端 Exactly-Once
调优	分区、广播、序列化、AQE、资源配置
部署	YARN 模式、参数配置、监控 UI