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 pyspark3.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, headerTrue, inferSchemaTrue)看起来最省事但它有个隐藏成本为了推断类型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)))三层嵌套加上括号我写的时候自己都得数缩进。换成 SQLorders.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_colquantity, price_colunit_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), onproduct_id, howleft)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 缓存用对地方别乱persistcache()或者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_udfPython代码跑在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 实战里磨出来的。至于后续还能往哪走我自己的规划是把这套离线链路和结构化流接起来让日报能近实时更新另外把特征工程部分也迁过来和模型训练打通。不过那是另一个坑了先把离线这条线跑稳再说。 
长河君 · 2026/10/26 3:39:53 · 阅读 1.2k PySpark实战：从pandas内存墙到TB级ETL清洗与性能优化
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 pyspark3.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, headerTrue, inferSchemaTrue)看起来最省事但它有个隐藏成本为了推断类型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)))三层嵌套加上括号我写的时候自己都得数缩进。换成 SQLorders.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_colquantity, price_colunit_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), onproduct_id, howleft)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 缓存用对地方别乱persistcache()或者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_udfPython代码跑在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 实战里磨出来的。至于后续还能往哪走我自己的规划是把这套离线链路和结构化流接起来让日报能近实时更新另外把特征工程部分也迁过来和模型训练打通。不过那是另一个坑了先把离线这条线跑稳再说。