企业官网建设流程全解析

面试 Spark 相关的岗位，简历上写了“熟悉 Spark Core”，结果第一轮被问 RDD 的五大特性直接卡壳，这种场景我见得太多。Spark Core 的知识点看起来很散——RDD、依赖、DAG、Stage、Shuffle、内存模型、序列化、广播变量，一条条背下来像在背八股，但真正被追问三层之后就发现，这些东西其实是一根链条上的不同环节：RDD 决定了血缘，血缘决定了依赖，依赖决定了 Stage 怎么切，Stage 决定了 Task 怎么发，Task 跑起来又要吃内存和网络。你只要把这根链条捋顺，绝大多数面试题都能顺着推导出来，而不是硬记。这篇内容我按照面试官实际提问的顺序，把 Spark Core 的核心考点重新拆一遍，每一块都补上“为什么是这样设计”和“参数怎么算出来的”，同时给出可以直接照着跑的代码和排查路径。适合两类人看：一类是准备面试、需要把零散知识点串成体系的；另一类是已经在写 Spark 任务、但遇到 OOM、数据倾斜、Shuffle 失败只能靠改参数碰运气的。下面这些东西，都是我在实际项目和模拟面试里反复验证过的。

1. 面试官在 Spark Core 里到底想听什么

1.1 一道“送分题”为什么能刷掉一半人

“说说你对 RDD 的理解。”这道题几乎每场必问，但真正答得好的比例低得惊人。绝大多数人的回答停在“弹性分布式数据集、不可变、可分区、能并行计算”这几句，听起来完整，实际上只是把定义翻译了一遍。面试官接着问一句“那弹性体现在哪”，就答不上来了。

我更建议把 RDD 拆成“它解决了什么问题”来讲。MapReduce 时代，多个作业串联靠中间落盘，一个 10 步的流程要写 9 次磁盘，每次落盘都是全量序列化。RDD 的解法是：不存数据，存“怎么算出来的”这份血缘关系，中间结果尽量留在内存里，出错就按血缘重算。所谓“弹性”，指的是分区数量可以随数据规模调整、数据可以缓存在内存也可以落盘、某个分区丢了能靠血缘重建、Task 失败了能自动重试——这四点里任意一点都是可以展开两分钟的素材。

顺着这个思路答，面试官会认为你理解了设计动机，而不是背了文档。设计动机是区分“用过”和“懂”的分水岭，后面所有的追问——为什么不可变、为什么要惰性求值、为什么要有宽窄依赖——本质上都是同一个设计动机的延伸。

1.2 我把 Spark Core 的考点分成四层

为了不让自己复习的时候东一榔头西一棒子，我习惯把考点按“从下往上”分成四层，每层有明确的核心问题和常见追问深度：

这张表的价值在于：当你被问到一个具体问题时，先定位它在哪一层，再往上一层和下一层各推一步，回答自然就有厚度。比如被问 broadcast 变量，定位在第四层，往上一步说它避免了每个 Task 都带一份只读数据的开销，往下一步说它的实现是每个 Executor 存一份并共享给所有 Task，配合 BlockManager 拉取。三句话下去，深度就出来了。

2. RDD：整个 Core 的地基，别只会背五大特性

2.1 五大特性逐条拆，每条都对应一个面试追问

RDD 的五个内部特性，每一条后面都挂着一个高频追问，我把它们列在一起，方便对照记忆：

1. 一组分区（Partitions）。追问：分区数怎么来的？读文件时由 split 决定，默认等于 HDFS block 数；经过 shuffle 由spark.default.parallelism或spark.sql.shuffle.partitions决定；repartition可以显式改。追问延伸：分区数太少会怎样？资源利用率低，一个 Task 扛全量数据；太多呢？Task 调度开销和大量小文件写出的问题。
2. 每个分区上有一个计算函数（compute）。追问：这个函数是什么时候执行的？答案是不执行，直到遇到 Action。这就是惰性求值，它的价值在于让优化器（虽然 Core 层优化有限）有机会把多个窄依赖合并成一条流水线，避免中间结果物化。
3. 依赖关系（dependencies）。追问：宽依赖和窄依赖怎么判定？这条是重灾区，下一节单独说。
4. 可选的分区器（Partitioner）。追问：哪些算子会产生分区器？只有groupByKey、reduceByKey、combineByKey、partitionBy这类需要按 key 重分布的算子才会设置分区器。追问延伸：为什么reduceByKey比groupByKey好？因为前者先在 map 端做局部聚合（combiner），后者直接把同 key 全量数据拉到 reduce 端。
5. 首选位置（preferredLocations）。追问：这就是数据本地性的来源，读 HDFS 时返回 block 所在节点，调度器据此尽量把 Task 发到数据所在机器。

五条特性里，第四条和第五条最容易被忽略，但它们恰好是“为什么 Spark 快”的两个关键：分区器避免了不必要的 Shuffle（两个 RDD 分区器相同且分区数相同时，join 可以直接在本地做），首选位置减少了网络传输。

2.2 宽窄依赖那一刀，切出的是 Stage 也是性能

判断标准只有一句话：父 RDD 的一个分区是否只被子 RDD 的一个分区使用。是，就是窄依赖；一个父分区被多个子分区用，就是宽依赖。map、filter、union、mapPartitions、coalesce（不 shuffle 时）是窄的；groupByKey、reduceByKey、join（未预分区）、repartition、distinct是宽的。

这条判定规则的实际意义在执行层：遇到宽依赖就要切 Stage。因为宽依赖意味着数据要跨节点重组，父 Stage 必须先全部算完并落盘，子 Stage 才能开始拉数据。窄依赖则可以全部塞进同一个 Stage 里做成流水线，一条记录从头算到尾，不需要中间物化。

一个常见的坑是在面试里把coalesce一律说成窄依赖。准确说法是：coalesce(n, shuffle=false)是窄的，它只是把多个父分区合并到一个子分区，但coalesce(n, shuffle=true)底层走的是 shuffle，是宽依赖。这个细节能答出来，面试官基本会认为你不是只看了速成课。

另外一个值得主动提的点是容错代价。窄依赖下某个分区丢了，只需要重算对应的一个父分区；宽依赖下丢一个分区，可能要重算上游所有分区。这就是为什么宽依赖之后加 checkpoint 收益最大——血缘太长，重算成本会呈指数级上升。

2.3 cache 和 checkpoint 不是二选一，是配合用

很多人把这两个当成“缓存数据的两种方式”，其实定位完全不同。

cache/persist是把计算结果留在内存或磁盘，用来加速同一应用内的重复访问。它的存储级别有十几种，常用的就三个：MEMORY_ONLY（内存不够就丢分区，下次用重算）、MEMORY_AND_DISK（内存不够溢写到磁盘，适合数据量大于内存）、MEMORY_ONLY_SER（序列化后存，省内存但读的时候要反序列化，费 CPU）。血缘关系不会因为 cache 而断开，数据丢了照样能重算，这也是它足够可靠的原因。

checkpoint是把 RDD 真正写到可靠存储上（通常是 HDFS），并且切断血缘。它的价值不是加速，而是保命：当血缘长到几十上百步，一次失败重算的代价可能比整个作业还大。用 checkpoint 必须注意两点——第一，它是在当前 Job 算完之后另外起一个 Job来写数据的，所以最好先persist再checkpoint，否则前面那段会被算两遍；第二，checkpoint 目录必须提前用sc.setCheckpointDir设好。

实际写法大概是这样：

val sc = spark.sparkContext sc.setCheckpointDir("hdfs:///tmp/checkpoint-dir") val base = sc.textFile("hdfs:///data/input/*.log") .map(parseLine) .filter(_.isValid) // 血缘很长，先缓存再 checkpoint base.persist(StorageLevel.MEMORY_AND_DISK_SER) base.checkpoint() // 第一次 Action 会顺带触发 checkpoint 的写盘作业 val cnt = base.count()

注意：checkpoint之后第一个 Action 的耗时通常会明显变长，因为多了一个写盘的 Job。如果是在做压测，别把这个当成性能回退。

还有一个替代方案是localCheckpoint，它把数据写到 Executor 本地磁盘并切断血缘，不依赖外部存储，速度快得多，但 Executor 挂掉数据就没了，也不跨应用可用。

最后补充一个面试里经常被追问的细节：cache之后如果数据被驱逐，再访问时会重新计算，但不会重新 cache——除非你再次触发并命中缓存策略。这意味着在内存紧张的场景下，反复驱逐再计算可能比不缓存更慢，需要结合存储级别和实际内存余量判断。

3. DAG 与 Stage 划分：调度链路上每一环都能被追问

3.1 从一行代码到 Task 上 Executor：完整链路

这条链路我认为是 Spark Core 面试里最值得完整背下来的一段，因为它把所有概念串成了因果链：

RDD 算子→ 触发Action→SparkContext.runJob提交 →DAGScheduler划分 Stage → 每个 Stage 生成TaskSet→TaskScheduler把 TaskSet 交给SchedulerBackend→ 按资源与本地性发到Executor的ExecutorBackend→Executor里起TaskRunner线程池执行 → 结果回传Driver的TaskScheduler。

每个环节都至少有一个可追问点。提交阶段：spark-submit之后 Driver 在哪儿起？client 模式在提交机器上，cluster 模式在集群里某个节点上，client 模式下 Driver 和集群之间的网络抖动会直接导致作业失败。划分阶段：DAGScheduler是从最后一个 RDD反向遍历血缘，遇到宽依赖就切一刀，所以切出来的 Stage 是一个个ResultStage加一堆ShuffleMapStage。执行阶段：TaskRunner会做反序列化、调用run、把结果序列化回传，回传结果有大小限制（默认约 1GB，由spark.driver.maxResultSize控制），超了直接报错。

3.2 Stage 划分算法 + Task 数计算例题

划分算法本身不复杂，但面试官喜欢让你现场算 Task 数，这个必须练熟。规则只有一条：每个 Stage 的 Task 数等于该 Stage 最后一个 RDD 的分区数。

举个具体例子。假设从 HDFS 读一个 1GB 的 Parquet 文件，块大小 128MB，默认会切出 8 个分区：

val raw = sc.textFile("hdfs:///data/orders", 8) // 8 个分区 val kv = raw.map(l => (l.split(",")(0), 1)) // 窄依赖，仍 8 个 val agg = kv.reduceByKey(_ + _) // 宽依赖，默认并行度 val big = agg.filter(_._2 > 100) // 窄依赖 big.collect() // Action

这段代码会被切成两个 Stage。第一个 Stage（ShuffleMapStage）的 Task 数是 8，因为宽依赖前的最后一个 RDD 是kv，8 个分区。第二个 Stage（ResultStage）的 Task 数取决于 shuffle 后的分区数，由spark.default.parallelism决定，在 YARN 模式下默认等于max(executor 总核数, 2)。假设总共 20 个核，那 reduce 端就是 20 个 Task。总 Task 数是 8 + 20 = 28。

如果换成 DataFrame API，reduce 端分区数由spark.sql.shuffle.partitions控制，默认 200——这是很多小数据量作业莫名其妙的性能瓶颈来源：几万条数据也切 200 个 Task，每个 Task 只处理几百条，调度开销远大于计算开销。我一般的做法是按数据量估算，单个 Task 处理 128MB 左右的量比较合适：

目标分区数 ≈ max(shuffle 数据量 / 128MB, executor 总核数)

提示：reduceByKey、join这类算子在 DataFrame 里对应的是groupBy、join，它们的分区数走的是spark.sql.shuffle.partitions，和 RDD 的spark.default.parallelism不是一回事，调参别调错。

3.3 调度模式与本地性：决定“快不快”的隐藏项

调度模式有两个：FIFO和FAIR。默认 FIFO，先提交的 Job 先拿资源，后面的排队。问题在于如果一个长作业占满了所有核，后面的短作业（比如一个交互式查询）会一直等。FAIR模式下多个 Job 按权重分资源，短作业能更快拿到执行机会。开启方式是设spark.scheduler.mode=FAIR，然后在代码里用sc.setLocalProperty("spark.scheduler.pool", "pool_name")配合fairscheduler.xml划分池。

本地性级别从高到低是：PROCESS_LOCAL（数据在同一 Executor 内）、NODE_LOCAL（同一节点不同 Executor）、NO_PREF、RACK_LOCAL（同机架）、ANY（跨机架）。调度器的策略是先从最高级别试起，等spark.locality.wait（默认 3s）还没等到资源，就降一级。这个参数的取舍很典型：调大，数据本地性更好但等待更久；调小，Task 起来得快但可能要多走网络。我的经验是 Shuffle 之后的 Stage 可以把这个值调小甚至设成 0，因为 Shuffle 本身已经产生大量网络传输，本地性收益不明显；而读取 HDFS 的 Stage 可以适当调大。

4. Shuffle：Spark Core 面试的分水岭

4.1 Shuffle 为什么贵：从磁盘、网络、GC 三个角度算账

能说清楚“为什么贵”，才算真的懂 Shuffle。我从三个维度拆：

磁盘。map 端要先把结果按 key 分区写进内存缓冲，缓冲满了溢写（spill）成文件，最后 merge 成大文件。一个 map Task 如果 200MB 输出、缓冲 32KB 级别，可能产生几十上百次溢写，每次都是顺序写加最后的合并读。reduce 端要拉取属于自己的所有分区，也涉及大量的随机读和顺序写。

网络。reduce 端要从 N 个 map 节点拉数据，用spark.reducer.maxSizeInFlight（默认 48MB）控制单次请求量。如果 Shuffle 数据量 100GB，这 100GB 全部要过网络。这也是为什么“尽量在 map 端预聚合”是黄金法则——reduceByKey会把同节点同 key 的数据先合成一条，groupByKey不会。

GC 与内存。Spill 发生的前提是内存不够，而内存不够的根源往往是对象太多。100 万条(String, Int)的 Java 对象在堆上可能占几百 MB，序列化后可能只有几十 MB。这就是为什么内存优化里序列化永远排在第一位。

用一个具体的数算一下：假设 shuffle 写入 200GB，集群 100 个核，理论上一轮能处理 100 个 Task。如果每个 Task 平均 spill 5 次、每次写 100MB，磁盘写放大就是 5 倍。这个量级不看不知道，一看吓一跳，也解释了为什么单纯加 Executor 内存往往治不了 Shuffle 慢。

4.2 Hash Shuffle 到 Sort Shuffle 的演进

这条演进线是面试高频，因为它体现的是“用计算换存储”的思路转变。

未优化的 Hash Shuffle：每个 map Task 为每个 reduce 分区生成一个文件，文件数 = map Task 数 × reduce 分区数。1000 个 map、200 个 reduce，就是 20 万个文件。海量小文件导致文件句柄耗尽、磁盘随机写严重。

Consolidated Hash Shuffle：开一个spark.shuffle.consolidateFiles=true，同一个核上连续跑的多个 map Task 复用同一批文件，文件数降到 核数 × reduce 分区数。这是 1.6 之前的主流方案，但降低文件数仍然依赖 reduce 分区数，效果有限。

Sort Shuffle：彻底换思路。每个 map Task 只输出两个文件——一个数据文件、一个索引文件。数据按分区的 key 排序后连续写入，索引记录每个分区在数据文件里的偏移量。reduce 端根据索引直接定位读取。文件数从 M×R 直接降到 2M，彻底解决了小文件问题，代价是多了一次排序。

Sort Shuffle 内部还有优化：当 reduce 分区数小于spark.shuffle.sort.bypassMergeThreshold（默认 200）时，走 bypass 路径，跳过排序，直接按分区写临时文件再合并，省掉排序开销。这也是为什么小作业上加大分区数反而不见得快——它触发了排序路径。

4.3 Shuffle 参数调优表与实测踩坑

常用的 Shuffle 参数我整理成表，方便对照查：

踩过的坑里，最典型的是“盲目加大maxSizeInFlight”。这个参数本质是拿内存换网络请求数，一台 Executor 上同时跑 8 个 Task，每个都按 96MB 拉数据，瞬时内存占用就是 768MB，如果堆本身不宽裕，反而触发 GC 甚至 OOM。我的做法是先看 GC 时间占比，超过 10% 就先别动这个参数。

第二个坑是只调参数不改逻辑。曾经有个作业 Shuffle 写了 3TB，参数怎么调都上不去。后来发现 reduce 端用groupByKey把同一个 key 的上百万条记录全拉过来了，改成一个两阶段聚合，先随机打散做局部聚合再按 key 聚合，Shuffle 写量直接降到 400GB。参数调优的天花板是代码结构，结构不对，参数只能救急。

5. 内存管理、序列化与共享变量

5.1 统一内存管理模型与那几个关键比例

Spark 1.6 之后统一成一套模型：堆内内存分成三块——保留内存（默认 300MB，给系统用）、统一内存区、其他（用户数据结构、元数据）。统一内存区由spark.memory.fraction控制，默认 0.6；其中 storage（缓存）和 execution（shuffle 缓冲、join 哈希表等）共享同一个池，初始划分由spark.memory.storageFraction控制，默认 0.5。

这个模型的巧妙之处在于动态借用：execution 需要内存时可以抢占 storage 的空闲部分，storage 不够时也能要回来，只要对应的一方没有被占用。实际算一下：Executor 堆 8GB，那个统一内存区就是(8192 - 300) × 0.6 ≈ 4735MB，初始 storage 和 execution 各 2367MB。如果业务大量用persist，可以把storageFraction提到 0.6；如果 Shuffle 和 join 压力大，就往下压到 0.3 左右。

堆外内存是另一套参数：spark.memory.offHeap.enabled和spark.memory.offHeap.size。开启后 execution 可以走堆外，避开 JVM GC，适合缓存密集或 GC 时间长的场景。代价是不能用复杂对象，且需要自己控制总量。

5.2 序列化：Kryo 为什么快，什么时候不能换

序列化排在性能优化第一位，是因为它同时影响三件事：Shuffle 落盘大小、网络传输量、GC 压力。

Java 原生序列化会写入类的完整描述信息、字段名、类型元数据，一条(String, Int)可能序列化成几百字节。Kryo 不写这些，它依赖提前注册的类 ID，同一个对象通常只有原生序列化的 1/5 到 1/10。开启方式很简单：

val conf = new SparkConf() .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .set("spark.kryo.registrationRequired", "false")

但如果自定义类很多，强烈建议注册，否则 Kryo 会在运行时把类名写进去，效果打折：

conf.set("spark.kryo.classesToRegister", "com.example.Order,com.example.User") // 或者实现 KryoRegistrator conf.set("spark.kryo.registrator", "com.example.MyKryoRegistrator")

什么时候不能换？一是有第三方类不可序列化（比如某些 JDBC 连接对象），二是用了需要 Java 序列化的框架（部分老的 Hadoop 输入格式），三是自定义类里有 Kryo 处理不好的集合类型，容易踩到BufferOverflowException。后者可以通过调大spark.kryoserializer.buffer.max（默认 64MB）缓解，一般提到 256MB 就够。

5.3 广播变量与累加器：用错了就是事故

广播变量解决的是“大只读数据被每个 Task 复制一份”的问题。比如一张 50MB 的维表，1000 个 Task 各自带一份，就是 50GB 的网络传输和内存占用。用广播变量的正确姿势：

val dimMap: Map[String, String] = loadDimTable() // 在 Driver 上构建 val bc = sc.broadcast(dimMap) // 每个 Executor 存一份 val result = rdd.mapPartitions { iter => val local = bc.value // 每个分区取一次，别在 map 里取 iter.map(r => (r.id, local.getOrElse(r.key, "unknown"))) }

两个细节必须强调：bc.value要写在mapPartitions外面，写在里面就变成每条记录取一次，虽然有缓存但仍有开销；广播变量一旦创建就不能修改，需要更新只能销毁重建。另外 Driver 端构建广播数据本身会占内存，几 GB 级别的数据不适合广播。

累加器用来做全局计数或求和，常见用途是统计脏数据条数：

val badCount = sc.longAccumulator("badRecords") rdd.foreach { r => if (r.invalid) badCount.add(1) } println(badCount.value)

坑在于失败重算会导致重复累加。Task 失败重试、Stage 重算时，累加器可能被加多次，所以它只适合“近似统计”，不适合对账级别的精确计数。还有一点：只有在 Action 执行完之后读value才有意义，在 transformation 里读永远是初始值。

6. 数据倾斜与故障排查：面试最后一公里的加分项

6.1 倾斜的识别：先定位再动手

数据倾斜的表现很好认：绝大多数 Task 几十秒就跑完，剩下几个跑几十分钟甚至卡在 99% 不动。但“看出来”不等于“定位到”，定位要做三件事。

第一，看 Spark UI 的 Stage 详情页，把 Task 的 Shuffle Read Size 排序，如果某几个 Task 读了几十 GB 而其他只有几百 MB，就是 key 级别的倾斜。第二，把倾斜的 key 捞出来看，df.groupBy("key").count().orderBy(desc("count")).show(20)，看是不是某些默认值、空值或者业务上的“大客户”占了绝大多数。第三，确认倾斜发生在哪一侧——是 map 端写倾斜（某些分区数据多），还是 reduce 端读倾斜（某些 key 拉取量大），两者的解法不一样。

我遇到最多的是空值倾斜。左表几亿行里有几千万行关联键为 null，join 时所有 null 被分到同一个分区，那个 Task 直接卡死。这种情况优先考虑业务语义：如果 null 本来就不该参与关联，直接过滤掉，问题消失。

6.2 七种解法与各自的适用边界

我把常用手段按“改动成本”排了个序，实际用的时候从低往高试：

两阶段聚合的写法值得单独提一下，它是通用性最好的方案：

// 第一阶段：加随机前缀，把热点 key 打散 val salt = 100 val stage1 = rdd .map { case (k, v) => (s"${k}_${scala.util.Random.nextInt(salt)}", v) } .reduceByKey(_ + _) // 第二阶段：去掉前缀，再聚合一次 val stage2 = stage1 .map { case (k, v) => (k.split("_")(0), v) } .reduceByKey(_ + _)

代价是 Shuffle 数据量翻倍，所以 salt 不宜设太大，一般 10 到 100 之间，按单个 key 的数据量估：让打散后的每个子 key 落到 128MB 左右比较合适。

6.3 高频问题速查表

最后把面试里出现频率最高、又最容易答偏的问题整理成表，方便临场对照：

我在实际排查里最深的体会是：Spark Core 的绝大多数“性能问题”，根子都不在参数上。真正卡人的往往是三件事——分区数没按数据量设计、宽依赖之前缺少预聚合、序列化用的还是原生 Java。这三件事改完，参数调优才有意义；顺序反过来，就是在给一个结构错误的作业打补丁。另一个习惯是每次作业上线前先在 Spark UI 里把 Stage 的 Shuffle Write/Read 量和 Task 耗时的中位数、最大值记一笔，作为基线。后面一旦有“任务变慢了”的反馈，直接拿数据对比，比凭感觉猜快得多。最后分享一个小技巧：调参的时候一次只改一个，并且写清楚改前改后的指标，同一份配置反复横跳很容易把自己绕进去。