企业官网建设流程全解析

这套笔记最早的形态是二十几页随手记的 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 或 11，Spark 4 开始要求 Java 17）。很多人装完 import 报错，八成是 JAVA_HOME 没配或者版本不对。检查方法很简单，在 shell 里敲java -version，能打印出版本号就行。

第二种是下载官方预编译包，解压后配置SPARK_HOME和PATH。这种方式的好处是自带spark-submit、spark-shell等命令行工具，调试时能直接看 Web UI，排查问题方便很多。

第三种是容器化，把 Spark 和 Python 环境打在一个镜像里。团队协作场景下我更推荐这种，能彻底避免"我这里能跑你那里不行"的扯皮。不过本地学习阶段，第一种足够了。

2.2 SparkSession：一切的入口

Spark 2.0 之后，SparkSession统一了原来的SQLContext和HiveContext，成了唯一的入口对象。我习惯把它封装成一个小函数，方便在不同脚本里复用：

from pyspark.sql import SparkSession def build_spark(app_name="demo", shuffle_parts=200): 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降级成 RDD，RDD 可以通过spark.createDataFrame(rdd, schema)升级成 DataFrame，DataFrame 可以createOrReplaceTempView注册成临时视图然后用 SQL 查询。

在绝大多数业务场景里，我的默认选择是 DataFrame API。原因很实在：它有 schema 信息，Catalyst 能做优化；它的代码表达力足够覆盖 95% 的需求；它的执行效率通常比手写 RDD 高出一截，因为优化器会自动做很多你未必想得到的事情。

3.2 什么时候必须回到 RDD

RDD 不是过时的东西，有些场景绕不开它。我整理了几类必须用 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 时，inferSchema=True很方便，它会扫一遍数据自动推断每列类型。但我在生产脚本里几乎从不用它，一律显式声明 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, schema=order_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, truncate=False)这个truncate=False很重要，默认它会截断长字符串，你很可能因此看不到数据里的异常字符。再补一个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，能完全避免 shuffle：

from pyspark.sql.functions import broadcast user_dim = spark.read.parquet("path/to/user_dim") joined = df_clean.join(broadcast(user_dim), on="user_id", how="left")

广播的门槛由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 或 DataFrame，Python 侧可以用向量化操作处理整批，跨进程通信的次数从"行数"降到"批次数"，性能提升通常在三到十倍。

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.enabled=true，它会自动把过大的分区再切开。这个方案最省事，但只在 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 我常用的配置清单

下面这张表是我在本地和中小规模集群上反复验证过的一组配置，抄过去能解决大部分"跑得慢"的问题。注意这些值是起点不是终点，具体还要看数据形态。

序列化器这一项值得单独提一句。默认的 Java 序列化器体积大、速度慢，换成 Kryo 后网络传输量和序列化耗时都能明显下降。启用 Kryo 只需要两行配置，代价几乎为零，属于典型的低垂果实。

6. 常见报错与排查技巧实录

6.1 内存类报错：先分清是 driver 还是 executor

java.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。比如你在一个类里定义了处理函数，然后传给map，Spark 会尝试把整个对象实例序列化，如果这个实例里含数据库连接、文件句柄这类不可序列化的成员，就会报错。

解决办法是在函数内部创建这些资源，或者把需要的字段单独抽出来作为局部变量。我踩过最冤的一次是在闭包里引用了一个 lambda，而那个 lambda 引用了外层的 SparkSession，结果整个 session 都被尝试序列化。教训是：闭包里只放纯数据，别放运行时对象。

AnalysisException: cannot resolve column也是高频错误。原因通常是列名拼错、列名里有空格或特殊字符、或者经过 join 后列名产生了歧义。逐列打印df.columns看一眼，比盯着代码猜十分钟有效。

6.3 问题排查速查表

关于"两次结果不同"这一条我想多说两句。除了窗口排序键不唯一，还有一个原因是用了rand()这类非确定性函数做过滤或采样。这类函数在每次重算时会产生不同的随机序列，如果分区数变化或者发生重算，结果就不一致。要复现的话得用rand(seed=固定值)。

还有一个容易被忽略的点是first()和take(1)。它们返回的是"任意一个"元素，在多分区场景下不保证是排序后的第一个。想要确定的第一条，老老实实加orderBy再取。

最后再分享一个小技巧，也是我排查问题最常用的一招：在关键节点插入df.explain(True)，看优化后的物理执行计划。它会告诉你 Spark 到底选择了什么 join 策略、有没有做谓词下推、分区数是多少。很多时候你以为代码在按你想的方式执行，看了执行计划才发现完全不是那么回事——我至少有三次性能问题是在看到物理计划里那个BroadcastNestedLoopJoin的瞬间定位到的。