先放结论Spark不是用来替代SQL的而是用来替代那些单机处理不了、又不想写一堆MapReduce的分布式计算场景。我见过太多人一上来就背API结果连“作业为什么慢”“Executor为什么OOM”都说不清楚。这篇教程我会按自己实际跑生产链路的经验来写从定位、架构、环境、算子到调优、踩坑一次讲透。我是从接手一个每天处理几TB流量日志的离线任务开始系统用Spark的前后折腾了一年多。当时最痛苦的不是写不出逻辑而是写出来的作业要么跑不动、要么动不动就Executor丢失。后来我把Spark的源码层面机制和参数一个个试过来才慢慢摸清楚它的脾气。这篇文章就按我自己的学习路径组织没有教科书式的废话全部以“能跑、能调、能排查”为目标。1. Spark到底解决了什么问题先看清它的定位1.1 从MapReduce的痛点说起很多人不理解Spark为什么能“快”其实快是相对Hadoop MapReduce而言的。MapReduce把每个计算步骤拆成Map和Reduce两个阶段中间结果必须落到磁盘下一阶段再去读。你要是写一个多步骤的ETL每一步都要落盘一次I/O开销非常夸张。我记得第一次用MapReduce跑一个六步清洗任务单日数据量也就几十GB却跑了近两个小时大部分时间都耗在读盘和写盘上。Spark的思路是尽量把计算留在内存里中间结果能缓存就缓存能不落盘就不落盘。它把一系列操作构建成一张DAG有向无环图在行动算子触发后统一调度执行。同样那个六步清洗任务换成Spark后首跑大概二十分钟后续因为数据能复用甚至更快。这个差距在数据量越大的场景下越明显。1.2 Spark的定位一个统一的计算引擎Spark不是数据库也不是存储系统它本身不存数据只是计算引擎。它能读HDFS、对象存储、本地文件、JDBC数据源也能读Kafka做流式处理然后计算结果写到各种目标。这意味着你原来存储在HDFS或数据仓库里的数据都可以直接用Spark来做批量计算、特征工程、报表加工。它还给了你一套统一的API支持Scala、Java、Python、SQL。很多团队是这种搭配离线ETL用Spark SQL复杂特征逻辑用PySpark实时链路用Structured Streaming一套框架覆盖大部分需求不用在多个引擎之间切换。这也是很多数据平台选它做离线计算底座的核心原因。1.3 哪种场景不要用Spark光知道Spark能做什么还不够还得知道它不适合什么。Spark定位是批量计算和准实时处理不是低延迟OLTP查询。你要是想做一个线上接口要求几百毫秒返回结果那应该用关系型数据库或者OLTP引擎而不是Spark。另外如果数据量就几百MB单机Pandas完全能处理引入Spark反而增加运维成本和调度开销。我做过的判断标准大概是单机内存能轻松装下且计算逻辑不复杂就不上Spark数据量上TB、需要分布式计算资源或者需要跑超过一小时的批处理才值得用Spark。学会判断边界能省掉很多不必要的复杂度。2. Spark的骨架Driver、Executor、RDD与Stage调度2.1 两种角色Driver和ExecutorSpark集群运行作业时角色划分其实很清晰。Driver是你的作业“大脑”负责解析代码、生成执行计划、调度任务、汇总结果。Executor是真正干活的“工人”分布在集群各个节点上负责执行具体的task并把结果返回给Driver。这个划分直接影响你后面的排错思路。Driver挂了整个作业失败Executor挂了Spark会尝试重新调度失败的任务但如果频繁Executor丢失就要考虑资源或内存设置。我在生产环境看到过不少案例Driver内存给太小日志直接显示“Driver堆内存溢出”连调度都完成不了。所以配置资源时Driver和Executor要分开考虑。2.2 RDD、DataFrame与Dataset的取舍很多人刚学Spark时被三个数据抽象搞晕我先给一个直观结论RDD是最底层的抽象DataFrame是带Schema的分布式表Dataset是类型安全的DataFrame主要在Scala/Java里体现价值。实际工作中Python开发者基本只用DataFrameScala/Gava开发者偶尔用Dataset做编译期检查RDD更多用于底层API或需要细粒度控制的地方。用PySpark写数据处理时我会默认用DataFrame。原因是Spark SQL的Catalyst优化器能对DataFrame做很多自动优化比如谓词下推、列剪枝而RDD算子不具备这些优化性能差距很大。有一段时间我帮同事review代码他用RDD的map/filter处理几亿行数据作业跑得很慢改成DataFrame后速度提了不少靠的就是Catalyst的优化能力。2.3 DAG与Stage一个作业在集群里到底怎么跑这是理解Spark性能的核心。你写的转换操作会被记录成血缘关系lineage构成DAG行动算子触发后Driver会把DAG拆分成多个StageStage之间以Shuffle为边界。Shuffle是什么呢我用一个快递分拣的例子来解释假设每个快递员手上有一些包裹分区数据现在需要按省份重新分组每个人必须把包裹送到对应省份的货架上这个过程就是Shuffle。Shuffle会把所有数据重新洗牌、跨节点传输开销很大。所以写Spark作业时一个朴素的优化思路就是尽可能减少Shuffle次数能一次算子搞定的不要拆成两步。窄依赖map、filter等和宽依赖groupBy、join等的区别就是判断是否需要Shuffle的关键。窄依赖每个分区只需从父RDD的少量分区获取数据宽依赖则要跨很多分区重新分配也是Stage划分的依据。理解这个机制后你看到Spark UI上的Stage列表时就能立刻判断哪些步骤贵、哪些便宜。3. 本地搭环境从下载Spark到首个作业跑通的完整流程3.1 安装前的版本选择开发学习阶段我建议直接在你自己的电脑上搭一套local模式的Spark环境没必要上来就搭集群。版本选择上我用的是Spark 3.x配Python 3.8文件的API变化不大但性能和改进明显好于旧版本。另外要注意Spark对Java版本有要求我用的是Java 8/11Spark 3.2默认支持Java 8/11/17但生产环境稳一点选8或11。还有一个易忽略的点Python 3.12等较新版本可能和某些PySpark版本存在兼容性问题。我的建议是用Python 3.8到3.10之间这期间踩坑最少。3.2 下载与启动Spark去Spark官网下载预编译的发行版一般选择“Pre-built for Apache Hadoop 3.2 and later”这个选项解压后放到指定目录。然后配置环境变量把SPARK_HOME/bin加入PATH。启动本地交互环境验证cd $SPARK_HOME ./bin/pyspark打开PySpark Shell后敲一下sc.version能输出版本号就说明环境没问题。你也可以在Shell里快速验证一个WordCountrdd sc.textFile(README.md) rdd.flatMap(lambda line: line.split( )) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a b) \ .take(10)这一步能让你直观体会到RDD的转换和行动操作是怎么配合的。3.3 写第一个独立脚本并用spark-submit提交交互式Shell适合验证思路生产作业往往是用独立脚本提交。我习惯用spark-submit统一提交。先写一个最简单的脚本from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WordCount) \ .getOrCreate() df spark.read.text(hdfs:///data/input/words.txt) words df.selectExpr(explode(split(value, )) as word) counts words.groupBy(word).count().orderBy(count, ascendingFalse) counts.show() spark.stop()提交命令如下$SPARK_HOME/bin/spark-submit \ --master local[2] \ --name WordCount \ wordcount.pylocal[2]表示本地用2个CPU核心来模拟并行执行。如果你想看Spark UI提交时给--master local[*]然后访问http://localhost:4040能看到作业的Stage、Executor情况这是个很好的学习工具。3.4 生产提交常用参数格式等你要提交到集群时参数会复杂很多。我贴一份生产环境常用的模板spark-submit \ --master yarn \ --deploy-mode cluster \ --name etl_daily_report \ --conf spark.executor.memory4g \ --conf spark.executor.cores2 \ --conf spark.executor.instances20 \ --conf spark.driver.memory2g \ --conf spark.sql.shuffle.partitions400 \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ s3://path/to/job.py这里每个参数都是有讲究的后面调优章节我再展开。先记住executor内存、executor核数、executor数量、driver内存、shuffle分区数是最常调的五个参数。4. 日常最常用的DataFrame算子这些能覆盖八成需求4.1 读取与写出数据源格式决定一半效率实际开发里DataFrame最常打交道的是读写。读CSV时我会指定schema而不是让它自动推断尤其是大文件自动推断会额外扫描一遍数据性能损失很大。from pyspark.sql.types import StructType, StructField, StringType, IntegerType schema StructType([ StructField(user_id, StringType(), True), StructField(event, StringType(), True), StructField(cnt, IntegerType(), True) ]) df spark.read \ .option(header, True) \ .schema(schema) \ .csv(s3://data/events/)写出时coalesce和repartition是最容易混淆的两个算子。coalesce会尝试在减少分区时不发生全量Shuffle适合把大量小分区合并成大文件repartition则一定会重新分区适合需要强制均衡的场景。我通常这样用输出到数据仓库前限定输出文件数量df.coalesce(8).write.mode(overwrite).parquet(s3://output/result.parquet)4.2 高频转换算子示例以下是我在数据清洗任务中几乎每天都会用到的算子配了实际用例select只选择需要的列上下游衔接时省内存。withColumn基于已有列生成新列。filter过滤条件行常配合col函数。groupByagg做分组聚合。join多表关联务必注意小表放左/右以及Join类型。dropDuplicates去重注意指定字段。from pyspark.sql.functions import col, sum, when result (df .filter(col(cnt) 0) .withColumn(is_high, when(col(cnt) 100, 1).otherwise(0)) .groupBy(user_id) .agg(sum(cnt).alias(total_cnt), sum(is_high).alias(high_cnt)) .orderBy(total_cnt, ascendingFalse))这里有个细节filter要尽量前置越早过滤数据后续算子处理的数据量越小。withColumn多次调用会生成新列但不会修改原DataFrame所以要用链式写法或重新赋值。4.3 窗口函数与UDF窗口函数在计算分组TopN、环比、累计值时非常有用。我用一个取每个用户最近一次事件的例子from pyspark.sql.window import Window from pyspark.sql.functions import row_number window_spec Window.partitionBy(user_id).orderBy(col(event_time).desc()) latest_df df.withColumn(rn, row_number().over(window_spec)) \ .filter(col(rn) 1) \ .drop(rn)UDF要注意性能。普通的Python UDF会有序列化和逐行调用的开销大数据量下比较慢。优先用内置函数和SQL表达式实在要自定义逻辑建议用Pandas UDFpandas_udf成批处理。4.4 与SQL的互转如果你的团队更习惯写SQLSpark完全支持。createOrReplaceTempView之后就能用spark.sql查询df.createOrReplaceTempView(events) spark.sql( SELECT user_id, COUNT(*) AS cnt FROM events WHERE cnt 0 GROUP BY user_id ).show()我个人的习惯是复杂逻辑优先用DataFrame算子链可读性差时转SQL两者结合效率最高。过度追求一种写法反而会让代码难维护。5. 提升性能的四个突破口内存、并行度、连接和缓存策略5.1 先看Spark UI再改参数不要一上来就盲目调参。第一步永远是看Spark UI的Stage页面确认瓶颈在哪个Stage、是计算慢还是Shuffle慢、是否有数据倾斜。我在排障时先关注三个指标某个Stage的输入数据量、Shuffle读写量、单个Task的最大耗时。如果所有Task时间都长说明数据量大或资源不足如果只有个别Task异常慢优先怀疑数据倾斜。5.2 Executor内存与并行度的关系Executor内存不能简单拍脑袋。一个Executor里能同时跑的Task数由spark.executor.cores决定。比如设置executor.cores2那么一个Executor内同时最多跑2个Task。每个Task能分到的内存大致是executor.memory / cores的某一部分还要扣除一些Overhead。如果Task需要的内存超过这个配额就会触发GC频繁或直接OOM。我常用这套估算思路假设处理10GB数据并行度为200每个Task大概处理50MB那每个Executor给4~8GB一般够用如果数据分布不均匀某个Task可能吃几百MB甚至更多这时要检查是否有倾斜问题而不是盲目加内存。5.3 shuffle分区数最常被忽视的参数spark.sql.shuffle.partitions默认是200但这个默认值在小作业和大作业下都容易不合适。小作业分区太多浪费资源大作业分区太少导致单个Task处理数据量过大。我的经验是以每个Task处理100~500MB数据为参考来设分区数。比如你Shuffle后要处理20GB数据分区数设在40~200之间比较合理。数据量波动比较大的任务我习惯把分区数统一设为Executor总核数的2~3倍。如果使用Spark 3.x可以考虑开启自适应查询执行--conf spark.sql.adaptive.enabledtrue它会根据运行情况动态调整reducer分区数和Join策略省去不少手动调优。5.4 Join优化离不开广播变量Join是Spark里最贵的操作之一也是最容易出性能问题的环节。小表Join大表时强烈建议用广播。广播会把小表发送到每个Executor节点避免Shuffle。触发广播的条件是表大小小于spark.sql.autoBroadcastJoinThreshold默认10MB。你可以手动调大或直接用Hintfrom pyspark.sql.functions import broadcast result big_df.join(broadcast(small_df), key, left)最适合广播的就是维度表比如几百MB以内的客户表、城市表。超大表之间Join必须Shuffle这时尽量让两张表的数据在Join前就按相同规则分区能减少Shuffle量。5.5 persist与checkpoint的取舍如果你要在同一个作业里多次复用同一个DataFrame就该考虑persist而不是每次都重新计算。默认StorageLevel是MEMORY_AND_DISK内存不够会溢写到磁盘。如果你明确知道某个DataFrame只需用一次那就不要persist白白占内存。checkpoint和persist不同它会把数据实际保存到可靠存储比如HDFS并切断血缘。血缘链太长时某个节点失败重算代价极大用checkpoint可以提前备份关键中间结果。它的代价是写盘开销大所以只在关键节点用不要每个步骤都checkpoint。6. 生产环境跑了一年之后必须记住的这些坑6.1 数据倾斜先分辨是“真倾斜”还是“资源不足”数据倾斜表现为某一个或某几个Task运行时间特别长其他Task早就跑完了整个Stage等这几个Task。我遇到过一次一个分组聚合作业数据里有个“默认渠道”字段占了60%的数据量导致所有数据挤在同一个Task上。定位办法很简单打开Stage详情页看每个Task处理的Input和Shuffle Read数据量如果出现一个Task读了几GB而其他Task只读几MB基本可以判定是倾斜。6.2 数据倾斜解法一两阶段聚合两阶段聚合是应对“热点Key”的经典办法。思路是先给Key加一个随机前缀把它拆成多个Key做局部聚合再去掉前缀做全局聚合。from pyspark.sql.functions import concat, lit, rand # 第一阶段加随机前缀局部聚合 stage1 (df.withColumn(salt, (rand() * 10).cast(int)) .withColumn(salted_key, concat(col(key), lit(_), col(salt))) .groupBy(salted_key) .agg(sum(cnt).alias(partial_cnt))) # 第二阶段去掉前缀全局聚合 result (stage1.withColumn(key, regexp_replace(col(salted_key), _\\d$, )) .groupBy(key) .agg(sum(partial_cnt).alias(total_cnt)))随机前缀的粒度要控制好。前缀太少效果不明显前缀太多小文件变多。一般按热点程度取10~100即可。6.3 数据倾斜解法二广播小表与过滤倾斜Key如果倾斜是因为某个Key关联了一个巨大的小表可以考虑把这个小表广播掉如果倾斜Key本身就是脏数据、无意义数据比如空字符串、Null、默认值优先在计算前过滤掉。有一回我们链路里一个“未登录”用户占了一半数据直接把这类用户单独处理整体作业时间降了40%。还有一个思路是“先过滤、再聚合、后合并”核心思想是分别处理正常Key和热点Key最后union。6.4 Executor OOM大多数时候不是内存不够很多人遇到OOM第一反应是加内存但我发现大多数时候OOM的根因不是内存总量不够而是某个Task的中间数据暴涨比如groupBy后某个Key的数据量特别大或者collect把全量结果拉到Driver。前者按数据倾斜处理后者要避免对超大DataFrame使用collect改用take取抽样。另外当Executor上用了大量persist(MEMORY_ONLY)内存不够时会出现反复GC。这时不如改成MEMORY_AND_DISK把放不下的数据溢写磁盘比反复GC稳定得多。6.5 小文件问题平台性能的隐形杀手Spark写数据时如果并行度很高输出文件会很多可能每个文件只有几MB。下游读这些数据时打开文件、扫描元数据的开销非常大。我们之前写过一次任务输出18000个Parquet小文件下游Hive查询全表扫描跑了很久改成coalesce(200)后文件数量降到200个查询时间降了将近一倍。控制输出文件数的通用做法是写入前用coalesce减少分区无Shuffle或者用repartition按目标字段重新分区有Shuffle。如果写入的是分区表尽量按分区粒度控制比如每个分区一个文件。6.6 其他容易踩的“小毛病”序列化问题闭包里用了不支持序列化的对象作业在Driver端会报Task not serializable。解决办法是避免在闭包里使用外部非序列化对象或改用Kryo序列化。时区问题Spark默认按系统时区解析时间如果上游数据是UTC下游需要本地时间要在spark-session里配好spark.sql.session.timeZone否则时间字段会差8小时。Parquet schema兼容同一张表后来加了新字段用旧schema读会报错或丢弃字段最好用mergeSchematrue来兼容演进。Spark.sql.shuffle.partitions改完之后没效果确认是不是已经开了AQEspark.sql.adaptive.coalescePartitions.enabled会动态调整分区数手动参数可能被覆盖。6.7 最后一点运维经验把Spark UI纳入日常习惯我在每次作业跑完后都会顺手翻一下Spark UI看有没有Task卡顿、Shuffle量异常、GC时间过长的现象。这些日志级细节往往比跑一遍EXPLAIN更能看出来真实问题。平时开发阶段多用explain()看执行计划生产排查阶段多依赖Spark UI两者结合Spark作业基本跑得稳。这套打法跑了一年后我们离线链路的作业平均耗时降低了近一半Executor OOM和倾斜导致的失败基本绝迹。如果你正在学Spark或刚接手Spark作业我希望这篇教程能帮你少走几个月的弯路。按这个顺序去学先懂原理再上手碰到问题就知道往哪查。最后再分享一个小习惯每个作业跑完在Spark UI里截一张Stage耗时分布图存档积累一段时间后回看你会对自己的数据规律有更清晰的认识。