企业官网建设流程全解析

1. 从pandas的128MB内存墙到Spark的TB级输入：先想清楚你换工具的真实理由

去年冬天我接到一个需求，把某模拟电商平台的全年订单流水做成日报表，文件一共十七个CSV，解压后大概二十来GB。第一反应当然还是pandas，写了三十行代码，pd.read_csv一跑，进程直接被系统杀掉。换成chunksize分批读，聚合逻辑一复杂，中间结果又开始往内存里堆，最后算到一半报MemoryError，日志里连是哪一步挂的都没看清。那段时间我把"内存不够就加内存"的思路走到了死胡同，直到把同样的逻辑搬到PySpark上，代码几乎没有大改，只是把df.groupby换成了df.groupBy，把df['col']换成了F.col('col')，跑完只花了十来分钟。

PySpark 是 Apache Spark 的 Python 接口，它把数据的切分、调度、容错这些脏活交给了 JVM 侧的调度器，Python 进程只负责描述"我要算什么"。这个分工决定了它的核心价值：你的数据量超过单机内存、或者计算逻辑复杂到需要多轮 shuffle 的时候，它能让你用接近 pandas 的心智模型写出可扩展的作业。它适合的人群其实很明确：写过 pandas、懂groupby和merge、但被数据量卡住的数据分析同学；以及需要把离线报表、特征工程流水线固化成可调度任务的工程同学。如果你日常处理的数据从来没超过几百万行，说实话，pandas 加 DuckDB 的组合比 PySpark 舒服得多，启动一个 SparkSession 就要吃掉好几秒，纯属给自己找麻烦。

这篇文章我想按"学习笔记"的方式来写，不走 API 手册那种从头到尾罗列函数的路子。我更想把一个完整的案例拆开给你看：同一份模拟订单数据，从裸 CSV 到清洗后的分区 Parquet，中间涉及读、洗、关联、窗口、聚合、写出这一整条链路，每一环我为什么这么选、踩过什么坑、跑完之后怎么判断它是快还是慢。因为我自己学 PySpark 最大的困惑从来不是"这个函数怎么用"，文档一查就有；而是"这段代码为什么跑了四十分钟"和"这个报错到底在说哪一行"。

在正式动手之前，有一个认知必须先建立起来，否则后面所有的坑你都会理解错方向：PySpark 的 DataFrame 是惰性的，而且不可变。你写下的每一行filter、withColumn，都不会立刻执行，只是在内存里搭一棵逻辑计划树，直到遇到show()、count()、write这类"行动算子"才会真正触发计算。这件事的直接后果是——报错的位置和你写代码的位置往往对不上。你在第 30 行写的withColumn类型转换有错，可能等到第 80 行write的时候才炸出来，堆栈里指向的是一堆你看不懂的 Scala 类。我刚上手时被这个特性折磨了很久，后来养成一个习惯：每搭完一段逻辑就printSchema()看一眼，或者干脆.limit(10).show()强制触发一次，让错误尽早暴露。

另一个必须提前接受的事实是shuffle 是有代价的。pandas 里merge两张表，本质是内存里的哈希表查找，几百毫秒的事；Spark 里同样的join，要把两张表按 key 重新分发到不同节点，中间结果落磁盘、走网络，规模一大就是几分钟起步。所以你在 pandas 里随手写的五个join，搬到 Spark 上可能变成五个连续的 shuffle，耗时翻几十倍。理解这一点之后，你会开始有意识地把能合并的操作合并、能提前过滤的提前过滤、能广播的广播——这才是 PySpark 实战里真正拉开差距的地方，而不是背函数名。

2. 本地环境搭建：JDK、SPARK_HOME 与那几个必踩的版本坑

很多人学 PySpark 卡在第一步就不是因为代码，而是环境。我当时也折腾了小半天，所以这一节我把顺序理清楚，你照着走基本能一次过。

PySpark 的运行依赖 JVM，这是绕不开的前提。pip install pyspark只是装了 Python 侧的封装层和一份内置的 Spark 发行包，真正的执行引擎还是跑在 JVM 上的。所以机器上必须有可用的 JDK，推荐 8、11 或 17 这几个长期支持版本，别上太新的。装完 JDK 之后配好JAVA_HOME，这一步漏了的话你会在启动时看到JAVA_HOME is not set或者更隐晦的Py4JJavaError。

pip install pyspark==3.5.1 python -c "import pyspark; print(pyspark.__version__)"

版本这块我踩过两个坑，值得单独说。第一个坑是 Python 版本和 Spark 版本不匹配。Spark 3.5 之前对 Python 3.11 以上支持得并不好，你会遇到类似TypeError: an integer is required或者ValueError: b'...' is not a valid...这种完全看不出因果的报错。解决办法很朴素：把 Python 降到 3.9 或 3.10，或者把 Spark 升到明确声明支持的版本，别硬扛。第二个坑在 Windows 上，读本地文件时会报java.io.IOException: Could not locate Hadoop executable，原因是缺少 Hadoop 的本地运行时工具，需要额外配置HADOOP_HOME指向一个包含相应可执行文件的目录。如果你用的是 Linux 或 macOS，这一步基本可以跳过。实在不想折腾环境的，用容器镜像跑是最省事的，镜像里该有的都有。

环境通了之后，第一个脚本我建议就写最小可运行的版本，别一上来就搞集群配置：

from pyspark.sql import SparkSession from pyspark.sql import functions as F spark = (SparkSession.builder .appName("order_etl_demo") .master("local[*]") .config("spark.driver.memory", "4g") .config("spark.sql.shuffle.partitions", "8") .config("spark.sql.adaptive.enabled", "true") .config("spark.sql.adaptive.coalescePartitions.enabled", "true") .getOrCreate()) spark.sparkContext.setLogLevel("WARN") print(spark.version)

这里几个参数我解释一下为什么这么设，因为它们直接决定你本地跑作业的体感。master("local[*]")表示用本机所有可用核心起一个伪分布式环境，*会按你的 CPU 核数分配，一般不用改。spark.driver.memory是驱动进程的堆内存，默认只有 1g，你稍微处理大一点的数据就会 OOM，本地调试给到 4g 比较舒服。spark.sql.shuffle.partitions默认是 200，这个值是为集群规模设计的，你在本地八核机器上跑几十万行数据，硬生生切成 200 个分区，每个分区就几百行，任务调度开销远大于计算本身，跑起来慢得让人怀疑人生——这是新手最容易忽略、收益又最直接的一个调优点。

spark.sql.adaptive.enabled打开的是自适应查询执行，它会在运行时根据实际的数据统计信息动态合并小分区、优化 join 策略。这个特性从 Spark 3.0 开始默认开启，但很多老教程里没有，你照着抄配置的时候可能会被显式关掉，导致运行效率差一大截。本地学习阶段建议显式写上，省得被旧代码带偏。

还有一个入口级的小细节：同一个 Python 进程里getOrCreate()只会创建一个 SparkSession，你在 notebook 里反复执行建 session 的单元格不会重复创建，但如果中途改了配置想生效，需要先spark.stop()再重新建。我在调试参数的时候经常忘了这点，改了shuffle.partitions发现毫无变化，排查半天才想起来是旧 session 还在。养成习惯：改配置前先 stop，或者干脆重启内核，能省下大量无效排查时间。

顺便说下怎么确认环境真的能跑分布式逻辑，而不是"看起来能跑"。跑完上面那段之后，浏览器打开localhost:4040，能看到 Spark UI。这个页面后面排错会反复用到，现在先记住它在哪。如果 4040 打不开，说明 session 没起来或者端口被占了，Spark 会自动往后找 4041、4042，看日志里的实际地址。

3. 一份模拟订单数据的端到端清洗：从裸CSV到按日分区Parquet

这一节是全文的主体，我把整个 ETL 链路摊开讲。数据我用一份完全虚构的模拟订单表来演示，字段大概是订单号、用户号、商品号、数量、单价、状态、创建时间，故意留了些脏数据进去——空值、重复、类型错乱、时间格式不统一，因为真实业务数据从来不长文档里那个样子。

3.1 读文件时显式声明 Schema，别偷懒用 inferSchema

第一个决定就是 schema 怎么来。spark.read.csv(path, header=True, inferSchema=True)看起来最省事，但它有个隐藏成本：为了推断类型，Spark 会把整个文件完整扫一遍。二十GB 的数据，光推断就白跑一轮。更麻烦的是，如果某列前一万行都是整数、后面冒出来一个"N/A"，推断结果会退化成字符串，你后面所有数值计算全部报错，而且很难定位。

所以我现在的习惯是永远显式声明 schema，哪怕多写十几行：

from pyspark.sql.types import (StructType, StructField, StringType, IntegerType, DoubleType) order_schema = StructType([ StructField("order_id", StringType(), False), StructField("user_id", StringType(), True), StructField("product_id", StringType(), True), StructField("quantity", IntegerType(), True), StructField("unit_price", DoubleType(), True), StructField("status", StringType(), True), StructField("created_at", StringType(), True), ]) orders = (spark.read .option("header", True) .option("encoding", "utf-8") .schema(order_schema) .csv("data/orders/*.csv"))

注意我把created_at声明成字符串而不是直接TimestampType。原因是时间格式在源数据里经常不统一，直接按 timestamp 读，遇到格式不对的行整行会变成 null，而且是静默的，你根本不知道丢了多少。先按字符串读进来，再用to_timestamp指定格式转换，转换失败的行会变成 null，你可以单独统计、单独处理，数据质量才有可控性。这是我在数据对不上的时候被坑出来的经验——报表数字和业务方对不上，最后查出来是几千行时间格式异常被静默吞掉了。

order_id我设了False表示非空，这是 schema 层面的一种"契约"声明，但要注意Spark 读 CSV 时并不会真的强制校验，它更多是表达意图，实际的行级校验还得靠后面的 filter。别指望声明了非空就不会有 null。

3.2 清洗链路的顺序有讲究：先去重还是先过滤

拿到 DataFrame 之后就是清洗。这里有个容易被忽略的点：操作的先后顺序会显著影响执行效率，因为 Catalyst 优化器虽然聪明，但也不会帮你把所有逻辑重排。

clean = (orders .filter(F.col("order_id").isNotNull()) .dropDuplicates(["order_id"]) .filter(F.col("status").isin("paid", "shipped", "finished")) .withColumn("quantity", F.col("quantity").cast("int")) .withColumn("unit_price", F.col("unit_price").cast("double")) .filter(F.col("quantity") > 0) .withColumn("amount", F.round(F.col("quantity") * F.col("unit_price"), 2)) .withColumn("created_at", F.to_timestamp("created_at", "yyyy-MM-dd HH:mm:ss")) .filter(F.col("created_at").isNotNull()) .withColumn("dt", F.to_date("created_at")))

我把filter放在dropDuplicates前面，很多人会反过来写。为什么？去重需要 shuffle，把无关行先干掉再 shuffle，中间数据量会小很多。尤其是status过滤这种能把数据砍掉一半的条件，提前执行几乎是白赚的。

dropDuplicates(["order_id"])本身也是一个 shuffle 操作，它的语义是保留第一条。如果你需要"保留最新那条"，就不能用它，得用窗口函数按时间倒序取row_number() == 1，这个后面讲窗口的时候再说。

金额计算我用了F.round(..., 2)，这一步不是为了好看。浮点数在分布式环境下的聚合误差比单机更明显，sum出来的结果小数位能拖到十几位，写进数据库或者给人看都很别扭。在明细层就规整好精度，比在聚合层再补救要稳妥。

created_at转成 timestamp 之后我再filter(isNotNull())，把格式异常的脏行剔掉。这一步剔了多少行，我一般会单独count()一次记录下来，写进数据质量报告，别让脏数据悄悄消失。

3.3 写出格式与分区策略：为什么我几乎不再写CSV

清洗完就要落盘。Parquet 是我的默认选择，原因很实在：列式存储，压缩比高，而且带 schema，下次读的时候不用重新推断；更重要的是它支持分区裁剪——按日期分区之后，查询某一天的数据只会扫那一个目录，速度快一个量级。

(clean.write .mode("overwrite") .partitionBy("dt") .parquet("out/orders_clean"))

分区字段的选择上我踩过一个坑：分区粒度别太细。我一开始按小时分区，一天二十四小时，一年就是八千多个目录，写的时候产生海量小文件，元数据管理开销巨大，后续读的时候光列目录就要好几秒。改成按天分区之后舒服多了。经验值是单个分区目录下的文件控制在几百MB量级，小于这个数就该往上归并粒度。如果你的数据量不大，比如就几百万行，按天分区可能都嫌碎，按月或者干脆不分区更合适。

mode("overwrite")和mode("append")的区别也要心里有数。overwrite会先删目标目录，重跑任务时很方便；append则是往里加，重复调度两次数据就翻倍了。我在生产脚本里基本都用overwrite或者按日期overwrite单天目录，就是为了保证幂等——同一个任务跑十次和跑一次结果一样，这是离线任务最基础的要求，别指望靠人工记忆去避免重复跑。

写完之后有个动作我强烈建议养成：读回来验证一次。

check = spark.read.parquet("out/orders_clean") print("行数:", check.count()) check.select("dt", "amount").groupBy("dt").sum("amount").orderBy("dt").show(5)

这个习惯救过我很多次。有一次写入过程中某个分区任务失败但整体没报错（因为我当时没检查），读回来发现某天的数据是空的，如果直接交付就是事故。写完立刻校验行数、关键指标，比事后被人问"为什么这天的数字是零"要体面得多。

4. DataFrame API 还是 Spark SQL：两种写法的真实分界线

学 PySpark 的时候很容易纠结一个问题：到底该用df.groupBy().agg()这种函数式写法，还是直接spark.sql("SELECT ...")。我的答案是两个都要会，而且要在同一个项目里混用，因为它们各自有明确的优势区间，用错了会让代码要么难读要么难改。

4.1 什么时候该老老实实写SQL

多表关联加复杂条件判断的场景，SQL 的可读性碾压 API 链式调用。比如你要算一个订单归因，需要关联用户表拿注册渠道、关联商品表拿类目、还要按CASE WHEN分档，用 DataFrame API 写出来是这样的：

result = (orders .join(users, "user_id", "left") .join(products, "product_id", "left") .withColumn("channel", F.coalesce(F.col("channel"), F.lit("unknown"))) .withColumn("price_band", F.when(F.col("unit_price") < 50, "low") .when(F.col("unit_price") < 200, "mid") .otherwise("high")))

三层嵌套加上括号，我写的时候自己都得数缩进。换成 SQL：

orders.createOrReplaceTempView("orders") users.createOrReplaceTempView("users") products.createOrReplaceTempView("products") result = spark.sql(""" SELECT o.order_id, o.amount, o.dt, COALESCE(u.channel, 'unknown') AS channel, CASE WHEN o.unit_price < 50 THEN 'low' WHEN o.unit_price < 200 THEN 'mid' ELSE 'high' END AS price_band FROM orders o LEFT JOIN users u ON o.user_id = u.user_id LEFT JOIN products p ON o.product_id = p.product_id WHERE o.status IN ('paid', 'shipped', 'finished') """)

同样的逻辑，一眼能看完，业务方来 review 也能看懂。createOrReplaceTempView不落盘也不复制数据，只是给 DataFrame 挂个名字，开销几乎为零，这点很多人有误解，以为是物化视图。另外别忘了用完spark.catalog.dropTempView("orders")清理，不然同一个会话里重复注册会互相覆盖，调试时容易拿到上一次的旧数据。

4.2 什么时候函数式API更合适

反过来，需要复用中间结果、或者要把处理逻辑封装成函数的场景，API 更顺手。比如清洗逻辑我要在好几个脚本里共用，写成函数接收 DataFrame 返回 DataFrame 就很自然：

def normalize_amount(df, qty_col="quantity", price_col="unit_price"): return df.withColumn( "amount", F.round(F.col(qty_col).cast("double") * F.col(price_col).cast("double"), 2) )

SQL 里做同样的事得拼字符串，容易出注入和格式问题，也不好做单元测试。另一个场景是链式操作天然适合逐步调试，你可以在任意一环.limit(5).show()插进去看中间状态，SQL 就没这么方便了。

4.3 两种写法最终都进同一条优化管线

有个让我安心的事实：不管你是写 API 还是写 SQL，最终都会被 Catalyst 优化器转成同一套逻辑计划，再做同样的优化。谓词下推、列裁剪、常量折叠、join 重排序，这些优化对两种写法一视同仁。所以"SQL 比 API 慢"这种说法并不成立，选择依据纯粹是可读性和维护性。

想验证这一点，直接用explain()看执行计划：

result.explain("formatted")

输出里你会看到== Physical Plan ==，重点关注两个东西：有没有出现Exchange（这就是 shuffle，是性能杀手），扫描节点的ReadSchema里包含几列（列裁剪有没有生效）。我调试的时候经常对着这个计划看，"哦这里多了一次 Exchange，说明我那个 join 没被优化成广播"，然后针对性改。这个技能比背一百个函数都值钱。

5. 窗口函数与多表关联：两个最容易写错的地方

窗口函数是 PySpark 里威力最大也最容易误用的部分。它在 pandas 里对应groupby().transform()或者rolling，但 Spark 的窗口定义更显式，也更需要你理解它背后的数据分布。

5.1 窗口定义里少写partitionBy会发生什么

from pyspark.sql import Window w_user_time = Window.partitionBy("user_id").orderBy(F.col("created_at").asc()) first_order = (clean .withColumn("rn", F.row_number().over(w_user_time)) .filter(F.col("rn") == 1) .drop("rn"))

这段是取每个用户的首单，很典型的用法。关键在partitionBy("user_id")。如果你只写orderBy，窗口就变成了全局窗口，所有数据会被拉到一个分区里排序，几千万行数据直接压到单个 task 上，跑起来要么极慢要么直接 OOM。我见过不止一个同事在这里翻车，报错信息是Container killed by YARN for exceeding memory limits，完全看不出跟窗口有关。

另一个细节：row_number和rank、dense_rank在并列值上的行为不同。取首单用row_number是对的，即使同一秒有两单也会随机给一个序号，保证只留一条；如果你想保留所有并列的首单，就得用rank。这个区别在去重场景里很关键，选错了要么多留数据要么少留数据。

5.2 用lag算复购间隔：一个真实的分析需求

窗口还有个高频用法是算相邻行的差值，比如算用户两次下单的间隔天数，用来判断复购周期：

w = Window.partitionBy("user_id").orderBy("created_at") gap = (clean .withColumn("prev_dt", F.lag("dt").over(w)) .withColumn("gap_days", F.datediff(F.col("dt"), F.col("prev_dt"))) .filter(F.col("gap_days").isNotNull())) avg_gap = (gap.groupBy("user_id") .agg(F.avg("gap_days").alias("avg_gap_days"), F.count("*").alias("repurchase_cnt")))

lag取的是当前分区内按排序的前一行，第一行会得到 null，所以后面必须过滤或者用coalesce兜底，否则平均值会被 null 影响（Spark 的avg会自动跳过 null，但计数会算错，count("*")会把首单那次也算进去，这是两回事，得想清楚你要的是哪种口径）。

这里还有一个隐蔽的坑：同一天有多笔订单时，datediff算出来是 0，会拉低平均间隔。如果你的业务定义是"不同购买日的间隔"，需要先按user_id + dt去重再算窗口。这种口径问题不是技术层面的，但会直接导致你的分析结论是错的，在用窗口函数之前，先把"什么算一次购买"定义清楚，比调代码重要得多。

5.3 关联时该广播还是该shuffle

多表关联绕不开 join 策略的选择。规则其实很简单：小表广播，大表 shuffle。

from pyspark.sql.functions import broadcast enriched = orders.join(broadcast(products_small), on="product_id", how="left")

broadcast会把小表完整复制到每个 executor 的内存里，避免大表被重新分区，原本一次昂贵的 shuffle 直接省掉。判据是小表能不能塞进 executor 内存，一般几百MB以内都比较安全，我这里商品维表通常就几万行，闭眼广播。Spark 3.x 之后开启了自适应执行，默认广播阈值是 10MB，超过这个值它不会自动广播，所以你明知是小表但超过了阈值，就得手动加broadcast()提示。

反过来，如果两张表都是几千万行，广播会把 driver 撑爆，报OutOfMemoryError或者Total size of broadcasted table is too large。这时候老老实实让它 shuffle，然后关注spark.sql.shuffle.partitions的设置是否合理。

我一般会在 join 之后立刻看一眼执行计划确认策略生效：

enriched.explain()

看到BroadcastHashJoin就说明广播成功了，看到SortMergeJoin就是走了 shuffle 路线。这两种算子名字记住，排查性能问题的时候一眼能定位。

6. 从Spark UI反推性能瓶颈：shuffle、倾斜与缓存的三板斧

代码能跑通只是及格线，跑得快才是真本事。这一节讲我怎么定位和优化慢作业，流程完全是可复现的。

6.1 先看Spark UI，别凭感觉猜

作业慢的时候，第一件事是打开localhost:4040，进Stages页面。这里每个 stage 会列出耗时、输入输出大小、shuffle 读写量、以及每个 task 的耗时分布。

判断依据很直接：如果某个 stage 的 Shuffle Read 特别大，说明这里有一次重分区，是优化的主要目标。如果 task 耗时分布极不均匀——比如 199 个 task 都是 2 秒，剩下 1 个跑了 8 分钟——那基本就是数据倾斜，光调参数没用，得改数据分布。

我第一次看到这种分布的时候以为是集群问题，反复重跑还是那样，后来才发现某个热门商品的订单量是其他商品的上百倍，groupBy("product_id")之后所有热点全落在一个分区上。倾斜的本质是 key 的分布不均，不是代码写错了，所以解法必须从 key 入手。

6.2 数据倾斜的两种实用解法

第一种是加盐打散。给热点 key 拼一个随机后缀，把一条大 key 拆成 N 条小 key 并行算，最后再去掉后缀聚合一次：

SALT = 8 salted = (orders .withColumn("salt", (F.rand() * SALT).cast("int")) .withColumn("salted_key", F.concat_ws("_", F.col("product_id"), F.col("salt")))) partial = salted.groupBy("salted_key").agg(F.sum("amount").alias("part_sum")) final = (partial .withColumn("product_id", F.split("salted_key", "_")[0]) .groupBy("product_id").agg(F.sum("part_sum").alias("total_amount")))

代价是多了一轮聚合，但把原来一个跑八分钟的任务拆成八个跑一分钟的，整体收益很明显。盐值取多少要看倾斜程度，8 到 32 是比较常用的范围，太多反而增加小任务开销。

第二种更省事：如果倾斜发生在 join 上，优先考虑能不能改成广播。广播 join 不走 shuffle，自然就没有倾斜问题。我在实际项目里大概七成的倾斜问题都是靠"把大表 join 小表改成 broadcast"解决的，剩下三成才是真的两边都大，需要加盐。

6.3 缓存用对地方，别乱persist

cache()或者persist()能把中间结果留在内存里，避免重复计算。但它不是万能药，用错地方反而拖慢整体。

base = clean.filter(F.col("dt") >= "2024-01-01").cache() cnt_a = base.filter(F.col("status") == "paid").count() cnt_b = base.filter(F.col("status") == "shipped").count()

这种"一个基础集被多次复用"的场景，缓存收益很大，因为第一次计算完就留在内存，后面两次直接读。但如果一个 DataFrame 只用一次，cache()就是纯粹的内存浪费，还可能把真正需要缓存的数据挤出去。

还有个细节：cache()是惰性的，第一次行动算子触发时才真正填充。所以如果你 cache 完立刻count()一次做预热，后面的查询会明显变快。用完之后记得unpersist()，尤其是在长期运行的会话里，不然内存会一直涨。

6.4 AQE真的能救命

最后说一下自适应执行。它能在运行时根据实际 shuffle 后的分区大小自动合并小分区，还能把 sort-merge join 动态切换成 broadcast join。我遇到过好几次"明明是小表 join 但因为统计信息不准没走广播"，打开 AQE 之后自动就优化了。

.config("spark.sql.adaptive.enabled", "true") .config("spark.sql.adaptive.coalescePartitions.enabled", "true") .config("spark.sql.adaptive.skewJoin.enabled", "true")

最后那个skewJoin是专门处理倾斜的，它会把过大的分区自动拆开。开了之后我发现有些原本需要手动加盐的场景，现在不用改了。不过也别完全依赖它，AQE 是缓解不是根治，倾斜特别严重时还是得从业务逻辑层面处理。

7. UDF与pandas_udf：Python代码跑在JVM集群里的性能代价

PySpark 最诱人的地方是能用 Python 写逻辑，但这背后有一个代价很多人没意识到：你的 Python 函数是在 executor 上单独起进程跑的，数据要在 JVM 和 Python 进程之间来回序列化。这个开销在普通 UDF 上非常明显。

7.1 普通UDF为什么慢

from pyspark.sql.functions import udf from pyspark.sql.types import StringType @udf(StringType()) def classify(v): if v is None: return "unknown" return "high" if v > 200 else "low" df.withColumn("band", classify("unit_price"))

这段代码逻辑没问题，但执行时每一行都要序列化一次、传给 Python 进程、算完再序列化回来。一百万行就是一百万次跨进程往返，速度可能比纯 JVM 实现慢十倍以上。所以我的第一条原则是：能用内置函数解决的，绝不用 UDF。上面这段完全可以用F.when().otherwise()表达，性能差距是数量级的。

7.2 pandas_udf：用向量化把性能拉回来

确实需要自定义逻辑的时候，用pandas_udf，它基于 Arrow 做批量传输，一次传一批数据而不是一行，开销能降下来一个量级：

import pandas as pd from pyspark.sql.functions import pandas_udf from pyspark.sql.types import DoubleType @pandas_udf(DoubleType()) def robust_zscore(s: pd.Series) -> pd.Series: med = s.median() mad = (s - med).abs().median() if mad == 0: return pd.Series([0.0] * len(s)) return (s - med) / (1.4826 * mad) result = df.withColumn("rz", robust_zscore("amount"))

注意函数签名：输入输出都是pd.Series，长度必须一致，返回的 Series 长度对不上会直接报错。这个约束刚开始用容易忘，尤其是里面有过滤逻辑的时候。

还有一个前提：Arrow 必须可用，否则会报PyArrow is required。装一下pip install pyarrow就行。版本也有讲究，太老的 pyarrow 和新版 Spark 配合会出各种奇怪问题，建议装 10 以上的版本。

7.3 什么时候该彻底放弃Python UDF

有些场景我会直接放弃 Python 侧实现。比如需要极致性能的字符串处理、复杂的日期运算，我会优先找内置函数；再不行就考虑用 SQL 表达式拼出来，让 Catalyst 去优化。Python UDF 在 Spark 里是"逃生舱"，不是"常规武器"，这个定位想清楚了，很多性能问题自然就避开了。

顺带一提mapInPandas，它能让你对整个分区的 pandas DataFrame 做处理，适合那种"必须拿到整批数据才能算"的场景，比如按用户分组做时间序列拟合。代价是分区之间不共享数据，你的逻辑必须能按分区独立计算。

8. 几个让我少走弯路的排错习惯

写到这儿基本把链路走完了，最后分享几个纯经验性的东西，都是我在实际调试里攒下来的。

第一，报错先看Caused by最后一段。PySpark 的堆栈特别长，前面一大堆 Scala 类名，真正有用的信息在最后那几行。AnalysisException: cannot resolve 'xxx' given input columns这种就是列名写错了或者大小写不匹配；SparkException: Job aborted due to stage failure往下翻能找到一个具体 task 的异常，那才是根因。

第二，复杂逻辑一定拆步验证。我现在的习惯是每写完三段转换就.limit(100).toPandas()看一眼中间结果，确认字段、类型、空值分布都符合预期。因为惰性求值会让你在最后一步才发现前面某步错了，届时链条太长，根本定位不到是哪一环。中间验证的成本是几秒钟，排查的成本可能是半小时。

第三，小数据量的作业别用 Spark。这话我前面说过，但值得再说一遍。几千行数据、单机几秒能算完的事，用 Spark 启动就要十秒，加上调度开销，纯粹是自找麻烦。我的判断标准是：数据量超过单机内存的一半，或者需要和其他大数据集关联，才值得上 Spark。否则 pandas 加 DuckDB 的组合又快又省心。

第四，配置项集中管理。我早期写脚本喜欢在代码里到处.config()，后来改成统一从一个字典或者配置文件读，本地一套、测试一套，切换起来方便。shuffle.partitions这种参数在不同数据量下最合适的值差很多，硬编码在代码里迟早要改。

还有一个我最近才体会到的东西：先想清楚输出的粒度，再倒推代码结构。我一开始写 ETL 都是"先把数据洗干净再想怎么聚合"，结果洗完之后发现需要的维度对不上，又回头改清洗逻辑，来回折腾。后来改成先写目标表的字段和粒度（比如"每天每渠道的 GMV 和订单数"），再倒推需要保留哪些列、在哪个阶段聚合，返工率明显下降。这个思维方式跟用什么工具没关系，但确实是我从 PySpark 实战里磨出来的。

至于后续还能往哪走，我自己的规划是把这套离线链路和结构化流接起来，让日报能近实时更新，另外把特征工程部分也迁过来，和模型训练打通。不过那是另一个坑了，先把离线这条线跑稳再说。