Apache Spark 这几年在大数据计算里几乎成了标配。做日志清洗用它跑离线报表用它搞实时特征也离不开它。我在实际项目里用它处理过几十TB的日志数据也帮同事排查过各种 OOM 和任务卡死的问题。很多人觉得 Spark 上手容易无非是写几个 map、filter、reduceByKey但真到了调优和故障排查的时候如果不懂它底层的分区、内存、shuffle 机制往往就只能靠加资源和重启硬扛。这篇内容是我把这几年用 Spark 的实际经验整理出来的完整笔记。核心思路是先把 Spark 为什么快、快在哪里讲明白再带你把它跑起来最后落到性能调优和问题排查上。适合刚接触大数据的初学者也适合已经写过 Spark 作业、但遇到性能和稳定性问题不知道怎么定位的同学。内容不追求面面俱到重点放在你真正用得上的部分。1. 为什么需要Spark从MapReduce时代的痛点说起1.1 MapReduce的硬伤在哪里在大数据框架里Spark 不是最早出现的它是在 Hadoop MapReduce 的基础上成长起来的。MapReduce 的模型设计非常简洁把计算拆成 Map 和 Reduce 两个阶段一批数据算完写回分布式文件系统下一批再读出来继续算。这个设计在十年前是划时代的因为它是第一个让普通工程师也能写分布式程序的计算模型。但它有一个致命问题中间结果落盘。想象一下一个需要多轮迭代的算法比如 PageRank 或者 K-Means每一轮都要把计算结果写一次磁盘下一轮再读一次磁盘。数据量一旦上去磁盘 IO 就成了最大的瓶颈。我算过一笔账同样规模的集群跑十轮迭代的机器学习任务MapReduce 的时间大概有六成以上花在读盘和写盘上真正计算的时间反而只占一小部分。这也意味着数据越多、迭代越多MapReduce 的劣势越明显。除了 IO 问题MapReduce 的编程模型也过于底层层。两个数据集的 join 要手动写一堆 shuffle 逻辑想表达一个简单的过滤加分组操作代码量往往很长。对于业务需求快速变化的场景这种开发效率是撑不住的。1.2 Spark的思路把中间结果留在内存Spark 的核心改进思路非常直接——把中间结果尽量留在内存里。它通过 RDD 抽象把数据分成多个分区可以缓存在内存中DAG 调度器对作业进行优化尽量减少 shuffle 次数血缘机制lineage通过记录每个 RDD 的依赖关系在数据丢失时不用重算整个作业只需要从血缘链上重新计算丢失的分区。这三件事合在一起让 Spark 在迭代计算和交互式查询场景下的性能比 Hadoop MapReduce 有明显提升。这不是靠某一个花哨的特性而是从计算模型上把最耗时的磁盘往返砍掉了。举个例子同样是上面的十轮迭代算法Spark 只需要第一轮把数据读进内存之后每一轮都在内存里完成计算整体耗时可能只有 MapReduce 的几分之一。当然Spark 现在早已不只是“快一点的 MapReduce”。它演进出了 Spark SQL、Spark Streaming、MLlib、GraphX 等组件覆盖了数据工程、实时计算、机器学习、图计算等多个领域。所以现在谈 Spark谈的是一整个生态而不是单一的计算引擎。1.3 Spark生态全景它到底能做什么Spark 的生态组件各自有明确的定位我整理了一个简单的对照组件功能定位典型场景Spark SQL / DataFrame结构化数据处理类 SQL 查询ETL、报表、即席查询Spark Streaming / Structured Streaming流式数据处理实时监控、实时特征计算MLlib分布式机器学习算法库分类、聚类、协同过滤GraphX图计算引擎社交网络分析、推荐关系计算这些组件全部基于同一个核心引擎代码也高度统一。最方便的一点是同样一套 Spark SQL 代码把它放在批处理任务里和放在微批流任务里改动基本只涉及读取方式和参数配置。这意味着团队不需要维护多套计算框架一套 Spark 就能覆盖大部分需求。刚接触 Spark 的同学我建议先从 Spark SQL 入手它最容易理解也最贴近日常的数据处理需求。等你把 SQL 映射到 DataFrame API 的写法摸熟了再去看 RDD 底层的执行原理整个知识体系会顺畅很多。1.4 什么场景该用Spark什么场景不该用虽然 Spark 很强大但也不是万能钥匙。我见过不少团队把 Spark 用在并不合适的场景上结果又慢又贵。我的判断标准是这样的适合用 Spark 的场景首推大规模数据的批处理。比如每天定时跑的数仓 ETL、日志清洗、用户行为聚合这些任务动辄处理几亿条记录用传统单机工具根本跑不动Spark 的分区并行能力正好发挥价值。其次是需要多轮迭代或复杂数据流的计算。推荐算法里的模型训练、图计算、复杂的多表 join 聚合都属于这一类。Spark 的内存计算和 DAG 优化能显著减少中间步骤的代价。不适合用 Spark 的场景也很明确毫秒级低延迟的在线查询那种要求单个请求在一两百毫秒内返回的场景更适合用专门的 OLAP 引擎或搜索引擎而不是 Spark。还有 OLTP 事务处理比如下单、扣库存这类需要强一致性和事务保障的操作也别往 Spark 上堆。业务体量还很小的时候一台单机数据库能搞定的事情强行引入 Spark 反而增加运维成本。2. 核心概念逐个击破RDD、DataFrame 与执行流程2.1 RDD你要认识的第一个抽象RDD全称是 Resilient Distributed Dataset弹性分布式数据集。这个名字听起来很学术但本质就是一个分布式的数据集合。你可以把它想象成一个特别大的列表但底层被拆成了多个分区每个分区是一个数据块分散在集群的不同节点上。RDD 有三个关键特性你必须理解。第一是不可变性RDD 一旦创建就不能修改想变换数据只能通过算子生成新的 RDD这保证了并行计算时不会出现数据竞争。第二是分区性分区数量决定了并行度多少个分区就意味着有多少个可以同时计算的任务。第三是弹性这里的弹性主要体现在容错上某个分区数据因为节点故障丢失了可以通过血缘关系从上一个 RDD 重新计算出来不需要整份数据重来。为什么要先讲 RDD因为后面 DataFrame、Dataset 虽然提供了更高级的 API但底层物理执行时数据依然是以 RDD 的形式分布的。你写的每一条 SQL最终都会被转换成 RDD 的 transformations 去执行。理解 RDD是看透 Spark 执行计划的第一步。2.2 Transformations 与 Actions惰性求值是怎么运作的Spark 的操作分成两类转换算子Transformations和行动算子Actions。这个区分是理解 Spark 执行流程的关键很多人刚学时容易栽在这里。转换算子包括 map、filter、flatMap、reduceByKey、join 等等。它们的共同特点是惰性求值——Spark 在执行这些算子时并不会立刻读数据和计算只是记录了“从当前数据如何变换到新数据”的描述相当于在纸上画了张图纸还没开始动手盖楼。真正触发计算的是行动算子比如 count、collect、saveAsTextFile、show。当你调用一个行动算子Spark 才会把前面所有转换算子的图纸汇总起来构建一张有向无环图DAG然后真正开始执行。这个设计有什么好处最大的收益是合并操作。举个例子你要从一个文件里读数据先过滤掉空行再取出需要的字段最后统计数量。在惰性求值的机制下Spark 可以把过滤、取字段、统计合并到一个阶段里执行避免每个步骤都读写一次数据。另外如果数据在内存里已经缓存重算的成本也很低。2.3 DataFrame、Dataset 与 Spark SQLRDD 虽然灵活但有两个明显的痛点代码写起来啰嗦而且完全不知道数据的结构schema信息。比如你有一个装着用户信息的 RDD里面是什么字段、字段是什么类型Spark 一概不知只能把它当作一堆对象来处理。DataFrame 的出现改变了这个局面。它带上了 schema 信息每一列都有明确的类型和字段名你可以直接在 DataFrame 上用 SQL 查询。更重要的是Spark SQL 引入了一个 Catalyst 优化器它会自动对代码进行谓词下推、列裁剪等优化。讲一个具体例子。假设你要查一个用户表里年龄大于 30 的用户姓名如果直接读 RDD 再逐行过滤Spark 会把整张表的每一列都读出来再过滤。如果写成 DataFrame 的 SQL 查询Catalyst 优化器会做一个列裁剪只读取 name 和 age 两列再把过滤条件提前到数据源读取阶段减少无效 IO。这种优化在 RDD 编程里得靠你自己手动做而 Spark SQL 已经内置了。Dataset 和 DataFrame 的关系在字面理解上有点绕。在 Java 和 Scala 里Dataset 是有类型的泛型集合DataFrame 是 Dataset[Row]即行类型是 Row 的 Dataset。在 Python 里PySpark 的 DataFrame 基本就等价于带 schema 的数据集你不需要太纠结这个类型层级用 DataFrame API 就好。2.4 宽依赖与窄依赖shuffle 为什么是性能黑洞RDD 之间的依赖关系分为窄依赖和宽依赖这是理解 Spark 性能的又一把钥匙。窄依赖的意思是父 RDD 的每个分区最多只被子 RDD 的一个分区使用。比如 map、filter、union 这种操作每个输入分区只有一个输出分区数据不需要跨节点传输Spark 可以在同一个执行线程里直接处理这些连续的窄依赖操作形成一条流水线。宽依赖则相反父 RDD 的每个分区可能被子 RDD 的多个分区使用。典型操作有 groupByKey、reduceByKey、join 等。这类操作会产生 shuffle数据需要跨节点传输并且为了保证同一个 key 的数据进入同一个分区通常要先落盘再做网络传输。在实际调优时我最先关注的就是作业里有多少次 shuffle。一个大数据作业如果出现三到五次大 shuffle每一次都意味着全量数据的网络传输和磁盘读写。减少 shuffle 次数比单纯增加内存和 CPU 核数的效果更明显。2.5 运行架构Driver、Executor 与 Cluster ManagerSpark 应用提交后有一个进程负责指挥有一群进程负责干活。负责指挥的是 Driver它运行你的主程序代码构建 DAG把任务切分好分发到各个执行节点。负责干活的是 Executor每个 Executor 有多个 task slot并行执行计算任务执行结果返回给 Driver。Cluster Manager 则负责资源的分配与回收。Spark 支持多种资源管理方式Standalone 模式用 Spark 自带的资源调度YARN 模式把 Spark 作业放到 Hadoop 的资源管理系统里统一调度近期很多团队也把 Spark 跑在 Kubernetes 上以容器作为资源粒度。不同模式下的 Driver 和 Executor 启动方式略有区别但应用代码在 API 层面基本是一样的。这里要特别提醒一点Driver 虽然是“指挥员”但它并不直接计算海量数据。如果你在代码里写了一个 collect() 把全量数据拉到 Driver 内存里Driver 很容易反正因为内存溢出而挂掉。这个错误新手特别容易犯后面我会详细讲怎么避免。3. 环境搭建与第一个 Spark 程序3.1 前期准备与运行模式选择搭建一个可用的 Spark 环境其实没有想象中复杂。先确认 JDK 版本Spark 3.x 版本通常要求 JDK 8/11/17具体版本要和发行包匹配。然后去 Apache 官网下载与你的集群环境匹配的预编译安装包解压到某目录配置好 JAVA_HOME、SPARK_HOME 环境变量即可。运行模式上有几种选择。本地模式local是最适合入门和调试的不需要任何集群Spark 在单机上模拟分布式执行。local[2] 表示用两个线程模拟两个执行节点local[*] 表示用所有可用的 CPU 核。本地模式学习足够但生产环境一般会用 YARN 或 Standalone。YARN 模式的好处是能和已有的 Hadoop 集群统一管理资源多团队共用时更容易分配。我这里建议学习阶段用 local 模式跑通逻辑部署阶段再切换到 YARN 或 Standalone这样两类问题不会纠缠在一起。3.2 用 PySpark 写第一个任务单词统计为了让你最直观地感受 Spark 的执行流程我用一个经典的单词统计任务做例子。这个任务虽然简单但包含了读取数据、转换、分组聚合、触发执行这几个最核心的环节。第一段代码用入门级的 RDD API 来写from pyspark import SparkConf, SparkContext conf SparkConf().setAppName(WordCountDemo).setMaster(local[2]) sc SparkContext(confconf) lines sc.textFile(data/input.txt) words lines.flatMap(lambda line: line.split( )) pairs words.map(lambda word: (word, 1)) counts pairs.reduceByKey(lambda a, b: a b) result counts.collect() for word, count in result: print(f{word}: {count})一段一段来解释。sc.textFile 从指定路径读取数据文件Spark 会按文件块大小决定分区数。flatMap 把每一行字符串拆成单词列表再把列表打平成一个单词流。map 把每个单词映射成 (word, 1) 的键值对。reduceByKey 把相同 key 的 value 累加起来得到每个单词的出现次数。最后 collect 是行动算子触发整个流程执行把结果收集到 Driver 进程里打印。Spark 2.0 以后更推荐用 SparkSession 入口它把 SparkContext、SQLContext、HiveContext 统一起来了。现代写法是from pyspark.sql import SparkSession spark SparkSession.builder.appName(WordCount).master(local[2]).getOrCreate()用这个写法你可以同样执行 DataFrame 版本的统计后面在上面的 DataFrame 层操作时也更方便。3.3 spark-submit 提交参数的讲究本地跑通只是第一步一个任务真正要在集群上运行通常会通过 spark-submit 命令提交。这个命令的常用参数我来按优先级说明。参数作用示例值--master指定集群管理器地址yarn / spark://host:7077--deploy-modeDriver 运行位置client / cluster--driver-memoryDriver 堆内存2g / 4g--executor-memory每个 Executor 堆内存4g / 8g--executor-cores每个 Executor 可用 CPU 核数2 / 4--num-executorsExecutor 数量10 / 50--conf任意 Spark 配置项spark.sql.shuffle.partitions400一个典型的提交命令长这样spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 2g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 20 \ --conf spark.sql.shuffle.partitions400 \ app.py这几个参数之间的关系很多人没想明白。executor-memory 乘以 num-executors 是你为这个作业申请的总堆内存而 executor-cores 乘以 num-executors 决定了总的并行 task 数。如果你申请了 20 个 Executor、每个 2 个核那么整个作业最多同时跑 40 个 task。分区数如果远高于 40任务会排队执行如果远低于 40部分核就闲着没事干。另外spark.sql.shuffle.partitions 默认是 200如果你的数据量很大这个数值偏低shuffle 时单个分区数据量过大会导致内存压力。这个参数在调优篇里还会再重点讲。3.4 本地调试常见的坑我在本地跑 Spark 时踩过不少坑这里挑几个最常见的提醒你。第一是文件路径。本地模式用 spark-submit 提交时文本文件的相对路径是基于当前工作目录的而不是代码文件所在目录。建议调试时候用绝对路径。第二是序列化问题。如果 map 算子内部引用了定义在外部的大型对象Spark 需要把这些对象序列化后下发到 Executor。对象不支持序列化时会报 SerializationException。解决办法是把需要的变量做成广播变量或者改成简单的基本类型。第三是分区数量。本地模式下默认分区数可能很少尤其是小文件数据跑出来的并行效果不明显。你会看到完成得快但 CPU 占用不高的现象这不代表程序有问题只是数据量和分区还没上去。4. 完整案例日志数据的清洗与用户行为统计4.1 案例背景与样本数据理论聊多了用实际案例串一遍更直观。假设某平台攒了一批用户访问日志每行记录包含时间、用户ID、访问URL、停留时长字段数据以竖线分隔。日志数量大约几亿条存储在分布式文件系统的某个目录下。任务有三个清洗无效记录比如字段数量不对、时长为负数的行统计每个用户的访问次数统计每天访问量最高的页面 Top10。样本数据长这样2024-06-01 10:00:00|user_001|/api/v1/home|120 2024-06-01 10:05:21|unknown|/api/v1/error|-1 2024-06-01 10:10:47|user_003|/api/v1/list?page2|320 2024-06-02 09:30:00|user_001|/api/v1/detail?itemId88|45可以看到第二行数据有问题用户ID是 unknown时长是 -1这种脏数据在真实日志里很常见。4.2 基于 DataFrame 的清洗与统计现在用 PySpark 的 DataFrame API 来写这个清洗统计流程。from pyspark.sql import SparkSession from pyspark.sql.functions import col, split, to_date, count, desc spark SparkSession.builder.appName(LogAnalyzer).getOrCreate() logs spark.read.text(hdfs://data/nginx_logs) # 按竖线切分字段并转换出需要的列 df logs.withColumn(time_str, split(value, \\|)[0]) \ .withColumn(user_id, split(value, \\|)[1]) \ .withColumn(url, split(value, \\|)[2]) \ .withColumn(duration, split(value, \\|)[3].cast(int)) # 清洗过滤字段缺失、时长非法、用户未知的数据 clean_df df.filter( col(user_id).isNotNull() col(user_id) ! unknown col(duration).isNotNull() (col(duration) 0) ) # 统计每个用户的访问次数 user_count clean_df.groupBy(user_id).agg(count(url).alias(cnt)) user_count.orderBy(desc(cnt)).show(10) # 每天访问量最高的页面 Top10 daily_hot clean_df.withColumn(day, to_date(time_str)) \ .groupBy(day, url) \ .agg(count(*).alias(pv)) \ .orderBy(desc(pv)) daily_hot.show(10)整个流程的逻辑很清晰读取原始日志拆分字段清洗脏数据然后做两种聚合。我在这个案例里特意没有写过于复杂的转换因为它的重点是让你看到 DataFrame API 在真实数据处理中的组织方式。groupBy 和 SQL 里的 GROUP BY 是一个意思agg 里可以放多个聚合函数同时算不同指标。orderBy 对应 SQL 的 ORDER BY。这些名词你看着应该眼熟实际上 Spark SQL 就是把熟悉的 SQL 能力搬到了分布式计算上。4.3 用 Spark SQL 的视角再看一遍同样的统计逻辑如果用 SQL 表达式来写会更贴近熟悉数据库的同学。把 DataFrame 注册成临时视图然后直接写 SQLclean_df.createOrReplaceTempView(clean_logs) sql_result spark.sql( SELECT user_id, COUNT(*) AS cnt FROM clean_logs GROUP BY user_id ORDER BY cnt DESC LIMIT 10 )你可能会疑惑这和我直接把 SQL 丢给 MySQL 有什么区别区别在于底层执行引擎不同。这张视图的数据是分布式存放在多个 Executor 里的GROUP BY 的聚合操作会被 Spark 自动拆分成 map 端的预聚合和 reduce 端的最终聚合数据量再大也能靠集群的分布式并行能力撑住。这就是 Spark SQL 和单机数据库本质上的不同。建议你在学习时同一套逻辑先用 DataFrame API 写一遍再用 SQL 写一遍对照着看执行结果。两周下来你对两种写法的等价关系会非常敏感日常也能根据场景选择更顺手的方式。4.4 运行时观察与验证执行完统计后不能只看结果数字对不对还要验证过程是否符合预期。如果集群上开着 Spark UI你提交这个作业后会看到好几个 stage。Stage 之间用宽依赖shuffle分隔每个 stage 的 input size 和 shuffle read/write 数据量都有记录。我在实际运行时特别关注两点一是清洗步骤过滤掉的行数占比如果比例偏高说明源数据的质量堪忧得反馈给上游二是 groupBy 之后的 shuffle 数据量如果某个 key 的行数特别多后面很可能出现数据倾斜。这些信号在刚执行完的时候就去 UI 上确认比事后排查高效得多。5. 性能调优从分区、内存到数据倾斜5.1 分区数量怎么定分区数量直接决定并行度。分区太少Executor 的 CPU 核利用不起来每个 task 要处理的数据量过大耗时变长分区太多task 数量会膨胀到几千甚至上万调度开销增加shuffle 时产生大量小文件下游任务读取会变得非常慢。怎么判断一个任务的分区数是否合理经验法则是每个分区的数据量建议控制在 128MB 以内并行度设定为总 CPU 核数的 2 到 3 倍。比如你有 20 个 Executor每个 2 核总共 40 核那么并行度设到 80 到 120 是一个比较稳妥的区间。如果算下来分区数远高于这个值优先用 coalesce 减少分区如果远低于这个值用 repartition 增加分区。coalesce 和 repartition 的区别容易混淆。coalesce 主要用于减少分区它会尽量不动现有数据只为合并部分分区而少量 shufflerepartition 则是等量重分布数据会引发全量 shuffle。能用 coalesce 减少分区时就别用 repartition因为它更节省 IO。5.2 内存参数到底怎么调Spark 的内存管理有一套核心比例参数很多人直接照抄网上配置却不知道每个参数的作用。我建议这么理解Executor 的堆内内存被分成三部分——执行内存、存储内存和预留内存。执行内存execution memory用于 shuffle、join、sort 等计算过程中的临时数据存储内存storage memory用于缓存 RDD 和 DataFrame预留内存是系统用的一小部分。两个关键比例参数是spark.memory.fraction执行内存和存储内存合计占堆内存的比例默认 0.6spark.memory.storageFraction存储内存占上述共享部分的比例默认 0.5。这两个值默认情况下大致让执行和存储各占 30%。如果你的任务里 shuffle 很重可以适当调低 storageFraction给执行内存更多空间如果你的任务主要是反复迭代读取同一份缓存数据则要反过来调高存储内存比例。另外别忘了堆外内存。Spark 的网络通信、Java 序列化等操作要用到堆外内存由 spark.executor.memoryOverhead 控制。在 YARN 模式下如果你申请的 executor-memory 是 4g实际每个 Executor 占用的容器内存通常还要加上 overhead 部分默认是 10% 或者最小值 384MB。这个参数配小了会出现 Executor 莫名其妙被 YARN 杀掉的情况。5.3 数据倾斜最让人头疼的性能瓶颈数据倾斜是 Spark 作业里最常见的性能杀手。现象很典型某个 stage 的大部分 task 几秒就完成了但有一个或几个 task 跑了几十分钟还在那里转圈整个作业卡着不动。原因往往很简单某个 key 的数据量远大于其他 key比如日志里某个爬虫用户贡献了全网 50% 的访问记录。定位数据倾斜的第一步不是看代码而是看 Spark UI。在 stage 页面里每个 task 的 duration 和 shuffle write size 会展示出来。如果某个 task 的 shuffle write 数据量是其他 task 的几十倍那基本就是它拖了后腿。处理方案要按场景选。如果倾斜只发生在单个超大 key 上可以先过滤掉这个 key 单独处理剩下的数据正常聚合最后再合并结果。如果是 join 操作导致的倾斜对小表使用广播变量Spark 会把它复制到每个 Executor完全避免 shuffle join。如果倾斜发生在 groupBy可以用加盐salting的方式给每个 key 拼上一个随机前缀拆成多个临时 key 并行聚合最后再去掉前缀汇总。加盐的思路我展开讲一下。假设你要统计每个用户访问量但某个头部用户的访问量独占 80%。直接把 user_id 作为分组 key那个头部用户的分区就会把单个 task 打爆。加盐的做法是读取数据时给这个头部用户的每行加入一个 0 到 9 的随机后缀这样同一用户的数据被分散到 10 个分区里各自统计最后再把这 10 个分区的结果相加。这个技巧简单有效也是所有 Spark 调优资料里都绕不开的核心手段。5.4 会用 Spark UI 是高级调优的分水岭很多有经验的同行在交流时第一句话就是你那个作业的 Spark UI 截图发我看一眼。原因在于Spark UI 把作业的真实运行状态完整呈现出来了不用靠猜。Stages 页面里每个 stage 的 input size / shuffle read / shuffle write / duration 是关键指标。如果一个 stage 的 shuffle write 数据量明显大于 input size说明这个 stage 产生了大量中间数据需要检查是过滤条件不够还是聚合粒度有问题。Executor 页面能看到每个 Executor 的 GC 时间GC 时间占比过高说明内存压力大优先考虑增大内存而不是加 CPU。SQL 页面会把每一条 SQL 对应的物理执行计划可视化你能看到哪些算子占用了大部分时间是不是出现了 BroadcastNestedLoopJoin 这类高代价操作。学会看 UI调优就成功了一半剩下的一半才是改参数。6. 高频问题与排查方法实录6.1 高频问题速查表我把这几年实际工作中遇到最多的 Spark 问题整理成了一张速查表现象可能原因排查起点常见解法Executor 抛出 OOM堆内存不足、分区不合理、缓存过多查看 Executor 日志与 Spark UI 内存指标增大 executor-memory增加分区清理无用缓存任务卡在某个 stage 超时数据倾斜、单条记录过大stage 内部 task 耗时分布shuffle write 数据量加盐、广播 join、单独处理倾斜 keyExecutor Lost Failure堆外内存不足、节点故障、心跳超时YARN 日志与系统监控增大 memoryOverhead检查节点资源调整心跳相关参数序列化异常自定义类型没有实现可序列化异常堆栈中的类名改为基本类型或实现 Java 序列化Driver 端内存溢出collect 拉取过多数据到 DriverDriver 日志避免 collect 全量数据用分区抽样代替shuffle file not found动态资源回收导致 Executor 被移除NodeManager 日志关闭过度激进的动态资源回收增加重试次数写入后产生大量小文件shuffle 分区数过高输出文件数与分区数的对应关系用 coalesce 合并分区控制 shuffle 分区数这张表解决的是“遇到问题往哪个方向猜”的问题但落到具体场景每一步还要更细地排查。下面挑两个典型的展开说。6.2 OOM 问题深挖先分清是哪里在溢出OOM 是 Spark 新手最容易遇到的错误但它的定位思路没你想的那么复杂关键是把阻碍分清是 Executor 的内存溢出还是 Driver 的内存溢出。Executor OOM 最典型的原因有三个。一是内存里缓存了太多数据尤其是用 cache/persist 缓存了整个 DataFrame却没有及时 unpersist存储内存被占满后执行内存也被抢占。二是某个分区内的数据量过大单条记录太大比如一个字段里存了几百 MB 的文本内容内存再大也装不下。三是 shuffle 过程中的聚合数据超过执行内存常见于 groupByKey 这种没有 map 端预聚合的算子。Driver OOM 的原因则比较单一绝大多数是代码里写了 collect()把分布式数据全部拉到 Driver 内存。处理方式有三种改用 take 只取部分数据、把结果保存到文件系统而不是拉回 Driver、把集合作业改成增量输出。另一个容易忽略的是堆外内存。如果日志里出现 Native memory exhausted 或者 Executor 被 YARN 判定为超过申请容器上限而杀掉多半是 memoryOverhead 配小了。这个参数在 Executor 内存接近上限时会显得特别重要。6.3 数据倾斜的定位与处理思路上一章提到数据倾斜的处理方案但没细说定位步骤。在真实项目里我通常按这个顺序排查第一步打开 Spark UI 的 stage 页面按 duration 排序看最长的几个 task。如果最长 task 的耗时是平均耗时的十倍以上高度怀疑倾斜。第二步看这个 stage 对应的算子类型。如果是 groupByKey 或 reduceByKey直接检查分组 key 的分布情况。最简单的办法是取少量数据在本地统计 key 对应的条数找出 top10 的 key。如果是 join看关联字段的分布。第三步回到代码里看是否有优化空间。优先检查几点是不是能用 map 端预聚合的 reduceByKey 而不是 groupByKey有没有需要在 join 前压缩数据量比如先过滤掉不需要的列小表能不能广播。我见过一个很典型的例子两个表做 join其中一个表只有两万行另一个表有两亿行。第一版代码直接用了普通的 shuffle join每次跑 40 分钟。改成广播小表之后整个作业缩短到 5 分钟以内。很多时候性能问题不是资源不够而是执行策略没选对。6.4 遇到没见过的问题怎么靠日志定位总会有一些错误不在你的经验范围内这时候排查思路比记忆更重要。我总结了一个三段式排查法实测对不熟悉的新错误很有效。第一段看异常类型和关键堆栈。不要从第一行开始读先找出带着具体异常类型的那一行比如 java.lang.NullPointerException 或 org.apache.spark.shuffle.FetchFailedException然后顺着它往下找你自己代码包的堆栈位置。第二段根据堆栈定位到具体算子类型。异常的堆栈里通常会体现是发生在 map 阶段、reduce 阶段还是 shuffle 阶段。如果你在 map 算子里的自定义函数抛出了异常堆栈里大概率能看到你函数的类名和行号。第三段对照官方文档与社区中的同类问题。Spark 作为成熟开源项目常见的坑几乎都有人踩过。搜的时候别搜整段报错搜关键异常类和涉及的模式名称就行。大部分情况下你会找到官方 JIRA 或社区帖子里给出同样问题的原因和解法。这一套流程走下来绝大多数问题都能在半小时内定位。真正棘手的是那些日志里没有任何异常、任务突然无声无息失败的场景那种情况多半要往资源竞争和网络稳定性方向去排查这就超出 Spark 本身属于集群运维的范畴了。最后说点实在话使用 Spark 这几年我最大的体会是大多数任务跑得慢或经常出错不是 Spark 本身的问题而是我们对数据分布、内存模型和分区策略的理解还没到位。刚上手的时候我习惯遇到问题就加 Executor 数量或堆内存后来发现效果往往有限。真正有效的是先把数据长什么样摸清楚再决定分区方案和参数配置。最后分享一个小技巧拿到一个陌生任务别急着优化参数先跑一个最小数据量的版本在 Spark UI 上看清楚每个 stage 的输入输出和耗时。数据规模小的时候定位问题成本最低等你把执行计划弄明白了再放大数据量就顺理成章。这套思路帮我省了不少排查时间也让我在团队里少背了好几个上线故障的锅希望对你有同样的帮助。