这套笔记最早的形态是二十几页随手记的 Markdown全是我在把一堆跑不动的 pandas 脚本往集群上搬的时候攒下来的。PySpark 的学习曲线很有意思语法看着和 pandas 差不多写起来也像那么回事但真正决定成败的东西全在语法之外——分区、shuffle、惰性求值、序列化这些词在单机世界里几乎不会出现在你的视野里。所以这篇 Python 学习笔记不走API 手册翻译的路子而是拿一份带脏数据的订单日志做案例实战从环境搭建一路走到调优和排错把每个关键选择背后的原因掰开讲清楚。如果你已经会写 Python、用过 pandas但一上大数据量就卡在内存和耗时上或者你正在准备数据方向的面试、想把 Spark 的实操链路真正走通一遍再或者你只是好奇 Spark 到底比 pandas 强在哪——这篇内容都适合你从头跟着敲一遍。我把代码、参数、踩坑记录全放进来了能直接抄的部分我尽量写成能跑的形式。1. 从单机 pandas 到集群 PySpark我为什么重写这套笔记1.1 一次内存爆掉的教训最早那份脚本干的活很简单读一天的订单明细 CSV按城市和品类做汇总输出一张报表。数据量小的时候跑得飞快直到某次全量回溯文件从几百兆变成十几个 G脚本在第 40 秒左右直接抛 MemoryError机器风扇狂转屏幕上的进度条再也没动过。当时的做法是切块读取、循环处理、最后合并勉强能跑但耗时从几分钟涨到两个多小时而且中间任何一步失败就得从头再来。换到 PySpark 之后同一份数据在同样的机器上本地单机模式local[*]跑完用了不到八分钟而且没有再出现内存溢出。这不是因为 Spark 的代码更聪明而是因为它的执行模型换了一套逻辑数据不再被一次性拽进 Python 进程的内存里而是被切成一个个分区在 executor 内部以迭代的方式流过算子。对于聚合类任务这意味着同一时刻真正驻留在内存里的只是当前分区的一小片数据。这个差异值得先讲清楚因为很多人学 PySpark 的第一步就走偏了——把 pandas 代码一行行翻译成 Spark 代码翻译完发现更慢然后得出结论Spark 不过如此。问题往往不在 Spark而在于你还在用单机的思路指挥一个分布式引擎。提示判断要不要上 PySpark先看数据量级和任务形态。几百兆以内、需要频繁随机访问中间结果的交互式分析pandas 依然是更舒服的选择数据量超过单机内存、或者需要跑批量定时任务、或者计算逻辑本身可以拆成 map/reduce 形态才是 Spark 的主场。1.2 PySpark 到底解决了什么问题把话说得直白一点PySpark 解决的是计算能力和存储能力不匹配的问题。单机时代算力和数据在同一台机器上数据涨了就得换更大的机器这叫纵向扩展成本曲线非常陡。Spark 换了个思路数据存在分布式的文件系统上计算也跟着数据走每个 executor 只处理离自己最近的那一份算完把结果汇总。这叫横向扩展加机器就能加算力。对写 Python 的人来说PySpark 的价值在于它把分布式编程的复杂度封装掉了。你不需要手动管理进程、不需要自己写任务调度、不需要处理节点故障重试写出来的代码形态和 pandas 很像底下那套 DAG 调度、shuffle 管理、容错重算的活它全包了。代价是你必须理解它的执行模型否则写出来的代码会在你看不见的地方疯狂搬运数据。我在笔记里记过一个很典型的对比同样一个 groupBy 聚合如果分区键选得对每个分区内部先做局部聚合再合并网络传输量能压到原来的十分之一如果分区键选错所有数据都要跨节点重分布耗时能差出七八倍。这个差别在代码上只体现为一个参数但背后的代价是实打实的。1.3 学习路径的取舍先跑通再调优我给自己定的学习顺序是先把一条完整的链路跑通读数据、清洗、聚合、写出再回头补原理最后才是调优。这个顺序很重要因为调优这件事极度依赖你对哪里慢的直觉而这个直觉只能从跑过真实任务里长出来。一上来就啃 Catalyst 优化器和 Tungsten 执行引擎看完还是不知道repartition和coalesce该用哪个。所以这份笔记的编排也是一样的逻辑第 2、3 节把环境和数据结构讲透第 4 节用一个完整案例走通全链路第 5 节专门讲性能第 6 节把常见的报错整理成速查表。2. 环境搭建与最小可运行骨架2.1 本地单机模式的三种装法新手最容易在环境上卡住。PySpark 的安装方式有三种常见路径我按上手难度排一下。第一种是纯 pip 安装一条命令搞定pip install pyspark。这种方式装完就能用但需要本机有可用的 Java 运行时Spark 3.x 一般要求 Java 8 或 11Spark 4 开始要求 Java 17。很多人装完 import 报错八成是 JAVA_HOME 没配或者版本不对。检查方法很简单在 shell 里敲java -version能打印出版本号就行。第二种是下载官方预编译包解压后配置SPARK_HOME和PATH。这种方式的好处是自带spark-submit、spark-shell等命令行工具调试时能直接看 Web UI排查问题方便很多。第三种是容器化把 Spark 和 Python 环境打在一个镜像里。团队协作场景下我更推荐这种能彻底避免我这里能跑你那里不行的扯皮。不过本地学习阶段第一种足够了。安装方式上手成本适合场景主要坑点pip 直接安装最低本地学习、小数据验证Java 版本不匹配、JAVA_HOME未设置预编译包 环境变量中等需要 Web UI 和命令行工具路径含空格、Python 版本混用容器镜像较高团队统一环境、CI 流程镜像体积大、数据卷挂载配置2.2 SparkSession一切的入口Spark 2.0 之后SparkSession统一了原来的SQLContext和HiveContext成了唯一的入口对象。我习惯把它封装成一个小函数方便在不同脚本里复用from pyspark.sql import SparkSession def build_spark(app_namedemo, shuffle_parts200): spark ( SparkSession.builder .appName(app_name) .master(local[*]) # 本地模式用满所有核心 .config(spark.sql.shuffle.partitions, shuffle_parts) .config(spark.sql.adaptive.enabled, true) .config(spark.sql.adaptive.coalescePartitions.enabled, true) .config(spark.sql.execution.arrow.pyspark.enabled, true) .getOrCreate() ) spark.sparkContext.setLogLevel(WARN) return spark这里有几个参数值得单独说。master(local[*])里的星号表示用本机所有可用核心本地测试时用它最省事但注意它和集群模式的语义完全不同——本地模式下 driver 和 executor 在同一个进程里你观察到的内存行为不能直接套用到集群。spark.sql.shuffle.partitions默认是 200在本地小数据上这个值偏大会生成一堆空任务拖慢速度跑小样本时我会临时调到 8 或 16。getOrCreate()这个方法名本身就说明了一件事同一个 JVM 里 SparkSession 是单例的重复调用不会新建而是返回已有实例。这个特性在 notebook 里很方便但也埋了个坑——如果你改了 config 再调getOrCreate()新配置不会生效因为实例已经存在了。改配置要用spark.conf.set(...)或者先spark.stop()。2.3 惰性求值与 DAG理解执行时机这是我建议所有从 pandas 转过来的人第一个要建立的概念。pandas 是即时执行你写df[df.a 1]这一行敲完数据就已经被过滤了。Spark 不是绝大多数算子select、filter、join、groupBy都是 transformation只负责往执行计划里追加一个节点不产生任何实际计算。真正触发计算的叫 action比如show()、count()、collect()、write。你调用 action 的那一刻Spark 才会把之前攒下的一堆 transformation 串成一张 DAG做优化然后切分成若干个 stage 提交执行。这个机制带来的第一个实际影响是同一份 DataFrame 被多次 action 使用时会重复计算。我早期写过一个脚本先用df.count()看了一眼行数然后又对df做了聚合结果整条链路跑了两遍白白多花了一倍时间。解决办法是加缓存df.cache()或df.persist()这个在第 5 节会详细展开。第二个影响是报错时机。你写了一百行 transformation代码跑得很顺利直到最后show()的那一刻才报错。别慌这不是你的代码没被执行而是本来就该在这一刻执行。理解这一点你在看堆栈信息时就不会困惑为什么错误定位到的是某个不起眼的 action 而不是出问题的那个算子。第三个影响是优化空间。因为 Spark 在真正执行前能拿到完整的执行计划它可以做一些你手写做不到的优化比如把 filter 下推到数据源、把相邻的 map 合并、调整 join 顺序。这也是为什么我更倾向用 DataFrame API 而不是 RDD——前者能吃到 Catalyst 优化器的红利后者不行。3. 核心数据结构选型RDD、DataFrame、Spark SQL 怎么选3.1 三者的分工与演进关系很多人被这三个概念绕晕其实它们的关系是层层递进的。RDD 是最底层的抽象本质是一个不可变的、可分区的记录集合你交给它的是一段段 Python 函数Spark 负责把这些函数分发到各个分区上执行。DataFrame 是在 RDD 之上加了一层 schema知道每列叫什么名字、是什么类型因此能做列裁剪、谓词下推这些优化。Spark SQL 则是 DataFrame 的另一种表达形式你用 SQL 字符串描述计算底层转成的还是同一套执行计划。我画不出图这类内容我一律用文字描述但你可以这样理解RDD 是一堆没有结构的对象DataFrame 是一张有表头的二维表Spark SQL 是用 SQL 语法操作这张表。三者的转换关系是DataFrame 可以.rdd降级成 RDDRDD 可以通过spark.createDataFrame(rdd, schema)升级成 DataFrameDataFrame 可以createOrReplaceTempView注册成临时视图然后用 SQL 查询。在绝大多数业务场景里我的默认选择是 DataFrame API。原因很实在它有 schema 信息Catalyst 能做优化它的代码表达力足够覆盖 95% 的需求它的执行效率通常比手写 RDD 高出一截因为优化器会自动做很多你未必想得到的事情。3.2 什么时候必须回到 RDDRDD 不是过时的东西有些场景绕不开它。我整理了几类必须用 RDD 的情况第一类是数据本身没有固定结构比如你要处理一批文本文件每行需要做复杂的字符串解析和自定义的状态维护。这类逻辑用 DataFrame 的 UDF 也能写但性能开销更大。第二类是需要对分区本身做精细控制。DataFrame API 层面你只能通过repartition和partitionBy间接影响分区而 RDD 提供了mapPartitions、mapPartitionsWithIndex这些算子能让你直接操作分区内部的迭代器。做批量外部调用比如每个分区建一个数据库连接时这个能力很关键。第三类是需要在节点间共享可变状态。accumulator和broadcast这两个机制在 RDD 层面用起来最直接。我给个实际例子。之前有个任务需要给几百万条记录逐条打一个自定义的相似度分逻辑写在一个第三方 Python 包里没法向量化。用 DataFrame 的 UDF 跑每条记录都要走一次 Python 序列化慢得离谱改成 RDD 的mapPartitions在分区内部先把模型加载一次然后批量处理速度快了大概四倍。def score_partition(rows): model load_model() # 每个分区只加载一次 for row in rows: yield (row[id], model.score(row[text])) result ( df.rdd .mapPartitions(score_partition) .toDF([id, score]) )注意mapPartitions里的函数必须能接受一个迭代器并返回一个迭代器用yield最自然。千万别在函数体里把整个分区list()出来塞进内存那等于把分区的意义废掉了。3.3 Schema 的力量为什么我坚持显式声明类型读 CSV 或 JSON 时inferSchemaTrue很方便它会扫一遍数据自动推断每列类型。但我在生产脚本里几乎从不用它一律显式声明 schema。理由有三个。第一是性能。推断 schema 需要额外扫一遍全量数据对大文件来说这是实打实的开销。第二是稳定性。推断逻辑对脏数据很敏感某一列混进了几个非数字字符整列可能被推成字符串下游的数值计算全部报错。第三是可控性你自己定义的类型能保证跨批次的字段名和类型完全一致不会因为某天数据分布变化导致 schema 漂移。from pyspark.sql.types import ( StructType, StructField, StringType, IntegerType, DoubleType, TimestampType ) order_schema StructType([ StructField(order_id, StringType(), False), StructField(user_id, StringType(), False), StructField(city, StringType(), True), StructField(category, StringType(), True), StructField(amount, DoubleType(), True), StructField(qty, IntegerType(), True), StructField(create_time, StringType(), True), ])显式 schema 还有个隐性好处它逼着你提前想清楚每列的业务含义。我在定义amount为DoubleType的时候就会问自己这列会不会有负值退款、会不会有空串、精度够不够。这些问题在写完 schema 之后基本都有答案了。4. 案例实战一份订单日志的完整清洗与统计链路4.1 数据源设计与模拟数据集为了让整条链路可复现我构造一个模拟的订单日志数据集。字段包括订单号、用户号、城市、品类、金额、数量、创建时间。数据里故意埋了几类脏数据完全重复的行、关键字段为空的记录、金额是字符串且带货币符号的、时间格式不统一的。raw_data [ (A001, u01, 杭州, 数码, 1299.00, 1, 2024-03-01 10:12:33), (A002, u02, 成都, 图书, 58.50, 3, 2024-03-01 11:02:10), (A002, u02, 成都, 图书, 58.50, 3, 2024-03-01 11:02:10), # 重复 (A003, None, 西安, 服饰, 299.00, 2, 2024-03-01 12:30:00), # 缺 user_id (A004, u04, 杭州, 数码, 899.00, 1, 2024/03/01 14:20:05), # 带符号格式不一 (A005, u05, 成都, 食品, None, 5, 2024-03-02 09:00:00), # 缺金额 ] df_raw spark.createDataFrame(raw_data, schemaorder_schema)这段代码本身没什么技术含量但我想说的是做案例实战一定要自己造脏数据。干净数据跑通只能证明语法没写错脏数据才能把when/otherwise、regexp_replace、na.fill这些真正干活的东西逼出来。4.2 读取与初步探查先看清楚再动手真实场景下数据一般在文件系统或数据湖上。读 CSV 的标准写法是这样df ( spark.read .option(header, true) .option(encoding, utf-8) .option(mode, PERMISSIVE) .schema(order_schema) .csv(path/to/orders/*.csv) )这里mode参数值得留意。PERMISSIVE是默认值遇到坏行会把整行塞进一个_corrupt_record字段不中断任务DROPMALFORMED直接丢掉坏行FAILFAST遇到坏行立刻报错。我的习惯是先按PERMISSIVE跑一遍看看有多少坏行、长什么样确认没问题再切到FAILFAST避免脏数据悄悄溜进下游。读进来之后第一件事永远是printSchema()和看几行样本。show(5, truncateFalse)这个truncateFalse很重要默认它会截断长字符串你很可能因此看不到数据里的异常字符。再补一个df.count()确认行数和源文件的行数对一下差额就是被丢弃的脏行。4.3 清洗空值、重复、格式、异常值四道关清洗这件事没有标准答案完全取决于业务口径。但有几道关是通用流程我按顺序讲。第一关是去重。dropDuplicates()不带参数会对全字段去重带上列名列表则按指定列去重。选哪种取决于你的业务定义——订单号唯一还是订单号商品唯一这得业务方说了算不是技术选型。df_dedup df_raw.dropDuplicates([order_id, user_id, create_time])第二关是空值处理。关键字段为空的记录我一般直接过滤掉因为补全的成本远高于丢掉这条样本的价值。df.na.drop(subset[user_id, amount])就是干这个的subset指定只检查这几列其他列有 null 不影响。第三关是格式统一这一步是坑最多的。金额字段混进货币符号、时间字段同时存在-和/两种分隔符这种问题在实际数据里极其常见。from pyspark.sql.functions import regexp_replace, trim, to_timestamp, col df_clean ( df_dedup .withColumn(amount, regexp_replace(col(amount), r[¥,\s], ).cast(double)) .withColumn(create_time, to_timestamp(regexp_replace(col(create_time), r/, -), yyyy-MM-dd HH:mm:ss)) .filter(col(amount).isNotNull() (col(amount) 0)) )这里解释几个细节。regexp_replace的第二个参数是 Java 正则\s表示空白符[¥,\s]这个字符组把所有干扰字符一次清掉。cast(double)遇到无法转换的值会返回 null 而不是报错所以后面必须紧跟一个isNotNull()的过滤否则你会拿到一堆 null 还以为是数据缺失。第四关是异常值。金额为负、数量为 0、时间戳在未来的记录都属于业务上不合理的数据。这类规则的写法就是普通的filter但规则本身需要和业务方确认我在笔记里专门留了一页记这些判断口径因为不同项目差别很大。4.4 聚合统计与多表关联把指标算准清洗完之后进入统计阶段。最基础的指标是按城市看订单量和成交额from pyspark.sql.functions import count, sum as _sum, round as _round, countDistinct city_stat ( df_clean .groupBy(city) .agg( count(order_id).alias(order_cnt), countDistinct(user_id).alias(user_cnt), _round(_sum(amount), 2).alias(gmv), ) .orderBy(col(gmv).desc()) )count(*)和count(order_id)的区别是个高频问题前者数行数null 也算后者只数非 null 的值。在已经清洗过关键字段的前提下两者结果一样但换成count(amount)就可能因为空值导致结果偏小。这种细节不写清楚报表对不上数是迟早的事。多表关联的关键是 join 策略的选择。小表 join 大表时用广播把小的那张表复制到每个 executor能完全避免 shufflefrom pyspark.sql.functions import broadcast user_dim spark.read.parquet(path/to/user_dim) joined df_clean.join(broadcast(user_dim), onuser_id, howleft)广播的门槛由spark.sql.autoBroadcastJoinThreshold控制默认 10MB。超过这个值的表广播会失败或者把 executor 内存撑爆所以我在代码里会显式加broadcast()提示同时在配置里把这个阈值调到更保守的值让它老实走 sort-merge join。提示left join 之后如果右表有重复键结果行数会膨胀这是数据对不上最常见的元凶之一。我现在的习惯是 join 前先在右表上groupBy(key).count().filter(col(count) 1)看一眼有没有重复键有的话先做去重或者确认业务逻辑。4.5 窗口函数排行榜与同环比的正确姿势窗口函数是我认为 Spark 里性价比最高的一组功能能省掉大量自连接。假设要算每个城市里成交额前三的品类from pyspark.sql.window import Window from pyspark.sql.functions import row_number, rank, dense_rank, lag w Window.partitionBy(city).orderBy(col(gmv).desc()) top3 ( city_category .withColumn(rn, row_number().over(w)) .filter(col(rn) 3) )row_number、rank、dense_rank三者的区别必须搞清楚面试高频实战也高频。假设成交额是 [100, 100, 90]row_number给出 1、2、3不管并列rank给出 1、1、3并列后跳号dense_rank给出 1、1、2并列后不跳号。要前三名通常用dense_rank要前三条记录才用row_number。同环比用lag或lead配合Window.partitionBy(维度).orderBy(时间)。这里有个细节如果你的时间列有重复同一天多条记录窗口的排序就不唯一结果会不稳定。我的做法是排序时补一个次级排序键比如orderBy([dt, order_id])保证每次跑出来的结果完全一致。这个坑我在不同项目里踩过至少两次。4.6 UDF 与 pandas_udf自定义逻辑的两种写法内置函数覆盖不到的逻辑就得自己写。PySpark 有两种 UDF普通 UDF 和 pandas UDF向量化 UDF性能差距非常大。普通 UDF 的工作方式是逐行序列化JVM 把一行数据传给 Python 进程Python 算完再传回去。每行都要走一遍这个跨进程通信开销惊人。我实测过一个简单的字符串处理普通 UDF 处理一千万行要四分钟左右。from pyspark.sql.functions import udf from pyspark.sql.types import StringType udf(StringType()) def mask_phone(p): if p is None or len(p) 7: return p return p[:3] **** p[-4:] df.withColumn(phone_masked, mask_phone(col(phone)))pandas UDF 换了个思路它不再逐行传而是按批次传一个 pandas Series 或 DataFramePython 侧可以用向量化操作处理整批跨进程通信的次数从行数降到批次数性能提升通常在三到十倍。import pandas as pd from pyspark.sql.functions import pandas_udf from pyspark.sql.types import DoubleType pandas_udf(DoubleType()) def weighted_score(amount: pd.Series, qty: pd.Series) - pd.Series: return (amount * 0.7 qty * 0.3).round(2) df.withColumn(score, weighted_score(col(amount), col(qty)))用 pandas UDF 有个前提spark.sql.execution.arrow.pyspark.enabled要设为true否则 Arrow 加速不生效。另外 pandas UDF 对 null 的处理和普通 UDF 不完全一致声明类型时建议用Float64Type可空而不是DoubleType避免遇到 null 直接抛异常。注意无论哪种 UDF只要用到了谓词下推和列裁剪的优化能力都会打折扣。能在内置函数里绕出来的逻辑就别写 UDF比如when/otherwise、regexp_extract、split、explode这些组合起来能解决很多问题。4.7 结果落地分区写出与文件组织最后一步是写出去。这里最关键的是分区策略。( result_df .repartition(64, dt) .write .mode(overwrite) .partitionBy(dt) .parquet(path/to/output/orders) )partitionBy(dt)会按日期把数据写成不同的子目录下游按日期查询时能直接跳过无关目录这叫分区裁剪partition pruning是最有效的优化手段之一。但分区列的选择要谨慎基数太高的列比如用户 ID会产生海量小目录把文件系统的元数据压垮基数太低的列比如性别又起不到过滤作用。日期、区域这类基数在几百到几千之间的列最合适。repartition(64, dt)这个写法是为了保证同一个 dt 的数据落在同一个分区里从而产生最少的文件数。如果直接repartition(64)再partitionBy每个目录下可能散落几十个小文件小文件问题会拖慢下游读取。mode参数也有讲究。overwrite会先删再写中途失败会导致数据丢失append适合增量场景但要自己做幂等errorifexists适合需要严格保证不覆盖的场景。生产上我倾向用overwrite配合写到临时目录再原子改名的方式用文件系统的原子操作换取安全性。5. 性能调优实录从跑不动到跑得快5.1 分区数量最容易被忽视的旋钮分区是所有 Spark 性能问题的根源。太少了单个任务处理的数据量过大容易 OOM 且并行度不足太多了每个任务的调度开销超过实际计算时间整体反而变慢。一个经验公式是目标分区数 ≈ 数据总量 / 每个分区理想处理量。每个分区理想处理量通常在 100MB 到 200MB 之间。假设你要处理 100GB 数据那么 500 到 1000 个分区比较合适。当然这只是起点还要看单条记录的复杂度。对于 join 和聚合执行过程中的 shuffle 分区数由spark.sql.shuffle.partitions控制默认 200。这个值在数据量变化时应该动态调整。Spark 3.x 引入了 AQE自适应查询执行开启之后它会根据实际的 shuffle 数据量自动合并小分区很大程度上免去了手工调参的痛苦。spark.conf.set(spark.sql.adaptive.enabled, true) spark.conf.set(spark.sql.adaptive.coalescePartitions.enabled, true) spark.conf.set(spark.sql.adaptive.advisoryPartitionSizeInBytes, 128m)开了 AQE 之后有个额外收益它能在运行时把 sort-merge join 自动切换成 broadcast join前提是检测到某一侧的数据量足够小。这个能力在数据分布不可预测的场景下非常有用。repartition和coalesce的区别也在这里讲清楚。repartition(n)会走全量 shuffle把数据重新均匀分布代价高但结果均匀coalesce(n)只是把相邻分区合并不走 shuffle代价低但只能减少分区数、且合并后的分区大小可能不均。要在写出前减少文件数用coalesce要在 join 前按某个键重新分布用repartition。5.2 数据倾斜分布式任务的头号杀手数据倾斜指的是某些分区的数据量远大于其他分区导致这些分区的任务成为整个 stage 的瓶颈。现象很好辨认Spark UI 里某个 stage 下大部分任务几秒钟就完成了就那么一两个任务跑了半小时还没结束。成因通常是 groupBy 或 join 的键分布不均。比如按城市统计某个一线城市的数据量可能是小城市的几百倍再比如按用户 ID join少数几个大客户贡献了大量记录。解决办法按代价从低到高排列第一种是加随机前缀打散。给倾斜键加一个随机后缀把一个大分区拆成若干个小分区聚合完再去掉后缀做二次聚合。这个方案通用于聚合场景。from pyspark.sql.functions import concat, lit, floor, rand SALT 8 df_salted df.withColumn(salted_key, concat(col(city), lit(_), (rand() * SALT).cast(int))) partial df_salted.groupBy(salted_key).agg(_sum(amount).alias(amt)) final partial.withColumn(city, split(col(salted_key), _)[0]).groupBy(city).agg(_sum(amt))第二种是广播小表。如果倾斜发生在 join 且有一侧是小表广播能彻底消除 shuffle也就消除了倾斜。这是最优解前提是表足够小。第三种是拆分倾斜键单独处理。把热点键的数据单独筛出来跑一遍剩下的走正常流程最后 union 起来。写法啰嗦但效果确切。第四种是开启 AQE 的倾斜 join 处理配置spark.sql.adaptive.skewJoin.enabledtrue它会自动把过大的分区再切开。这个方案最省事但只在 sort-merge join 场景下生效。5.3 缓存与持久化用对地方才叫优化cache()和persist()是双刃剑。用对了能把重复计算的时间砍掉一半以上用错了会把 executor 内存吃光、导致频繁 GC 甚至 OOM。判断标准很清晰同一个 DataFrame 是否会被多个 action 使用且它的计算代价是否足够高。两个条件都满足才值得缓存。如果只是show()两次看看结果缓存反而增加开销。存储级别的选择也有讲究。MEMORY_ONLY最快但内存不够时会丢弃分区下次用到要重算MEMORY_AND_DISK会溢出到磁盘牺牲速度换稳定DISK_ONLY最慢但最省内存。我在本地小数据下用默认的MEMORY_AND_DISK就够了集群上如果内存充足会显式指定MEMORY_ONLY。缓存用完之后记得unpersist()。这个习惯很重要尤其在同一个小集群里跑多个任务时忘记释放的缓存会一直占着内存直到 SparkSession 结束。我在笔记里专门标了红缓存不是越多越好是越精确越好。5.4 我常用的配置清单下面这张表是我在本地和中小规模集群上反复验证过的一组配置抄过去能解决大部分跑得慢的问题。注意这些值是起点不是终点具体还要看数据形态。配置项建议值作用说明spark.sql.adaptive.enabledtrue开启自适应执行自动合并小分区spark.sql.adaptive.skewJoin.enabledtrue自动处理 join 时的数据倾斜spark.sql.shuffle.partitions数据量GB数×2~4控制 shuffle 后的分区数spark.sql.autoBroadcastJoinThreshold30mb~50mb广播 join 的阈值上限spark.sql.execution.arrow.pyspark.enabledtrue启用 Arrow 加速 pandas UDFspark.serializerKryoSerializer更快的序列化器spark.sql.files.maxPartitionBytes128mb单分区读取的最大字节数序列化器这一项值得单独提一句。默认的 Java 序列化器体积大、速度慢换成 Kryo 后网络传输量和序列化耗时都能明显下降。启用 Kryo 只需要两行配置代价几乎为零属于典型的低垂果实。6. 常见报错与排查技巧实录6.1 内存类报错先分清是 driver 还是 executorjava.lang.OutOfMemoryError: Java heap space这个报错几乎每个写 Spark 的人都见过。关键是要先判断它发生在 driver 还是 executor两者的处理方式完全不同。如果错误堆栈里出现collect、toPandas、show这类 driver 侧的操作那问题在 driver。最常见的诱因是把大结果集 collect 回本地几千万行数据被拽进 driver 内存必然崩。解决办法是只 collect 聚合后的少量结果或者直接写成文件。如果堆栈里出现 shuffle 相关的类那大概率是 executor 侧内存不足。这时可以调spark.executor.memory但更根本的做法是减小单个分区的数据量——增加分区数、减少每个分区要缓存的量。单纯加内存往往只是把崩溃的时间点往后推。还有一个隐蔽的坑是spark.memory.fraction这个参数。它控制 executor 内存中用于执行和缓存的比例默认 0.6。缓存了大量数据时把执行内存挤没了shuffle 就会溢写到磁盘甚至 OOM。这种情况要么减少缓存要么调低这个值。6.2 序列化与类型类报错PicklingError和TypeError: cannot pickle ...这类问题在写 UDF 或用 RDD 时特别常见。根本原因是你闭包里引用的对象需要被序列化后发送到 executor而它本身不可序列化。最典型的场景是在类的方法里引用self。比如你在一个类里定义了处理函数然后传给mapSpark 会尝试把整个对象实例序列化如果这个实例里含数据库连接、文件句柄这类不可序列化的成员就会报错。解决办法是在函数内部创建这些资源或者把需要的字段单独抽出来作为局部变量。我踩过最冤的一次是在闭包里引用了一个 lambda而那个 lambda 引用了外层的 SparkSession结果整个 session 都被尝试序列化。教训是闭包里只放纯数据别放运行时对象。AnalysisException: cannot resolve column也是高频错误。原因通常是列名拼错、列名里有空格或特殊字符、或者经过 join 后列名产生了歧义。逐列打印df.columns看一眼比盯着代码猜十分钟有效。6.3 问题排查速查表现象可能原因排查与处理任务卡在某个 stage 不动数据倾斜或单分区过大看 Spark UI 的任务耗时分布对倾斜键加盐或广播小表结果行数比预期多join 右表键重复先对右表做键去重或确认业务口径结果行数比预期少隐式类型转换产生 null 被过滤检查 cast 后的 null 数量加isNotNull前先统计同一个脚本两次结果不同窗口排序键不唯一 / 未指定分区补次级排序键检查是否有非确定性函数写出后产生大量小文件分区数与输出目录不匹配写出前repartition到合理数量控制分区列基数缓存后反而更慢缓存了只需用一次的数据审查cache调用点用完及时unpersistUDF 执行极慢逐行序列化开销改用 pandas UDF 或替换为内置函数组合关于两次结果不同这一条我想多说两句。除了窗口排序键不唯一还有一个原因是用了rand()这类非确定性函数做过滤或采样。这类函数在每次重算时会产生不同的随机序列如果分区数变化或者发生重算结果就不一致。要复现的话得用rand(seed固定值)。还有一个容易被忽略的点是first()和take(1)。它们返回的是任意一个元素在多分区场景下不保证是排序后的第一个。想要确定的第一条老老实实加orderBy再取。最后再分享一个小技巧也是我排查问题最常用的一招在关键节点插入df.explain(True)看优化后的物理执行计划。它会告诉你 Spark 到底选择了什么 join 策略、有没有做谓词下推、分区数是多少。很多时候你以为代码在按你想的方式执行看了执行计划才发现完全不是那么回事——我至少有三次性能问题是在看到物理计划里那个BroadcastNestedLoopJoin的瞬间定位到的。