1. 为什么我把 PySpark 当成 Python 进阶的必修课单机 Pandas 处理几百万行数据就开始吃满内存、跑一个 groupby 要等好几分钟这大概是每个写 Python 数据分析的人早晚会撞上的墙。我最初也是抱着Python 数据处理够用了的心态直到一次需要清洗近两千万条日志记录Pandas 直接 MemoryError 崩在读取阶段才真正下决心把 PySpark 捡起来。这篇笔记就是那段时间的实战沉淀不是 API 手册翻译而是我从能跑通到能跑稳过程中整理出来的案例、参数和踩坑记录。先把话说清楚PySpark 是 Apache Spark 的 Python 接口它把分布式计算的调度、内存管理、容错这些脏活都封装在 JVM 侧你只需要用 Python 写转换逻辑。它能做到单机工具做不到的事——把 TB 级数据切分到几十台机器上并行处理而且代码写起来和 Pandas 有七八分相似。它解决的问题很直接数据量超出单机内存和算力上限时如何用可接受的成本完成清洗、聚合、分析和建模前的特征加工。这篇文章适合谁如果你已经会写 Python、用过 Pandas 的merge、groupby、apply但一遇到大数据就束手无策那这里的内容基本是为你准备的。如果你是完全的编程新手建议先把 Python 基础和 SQL 的join、group by弄明白再来否则 Spark 的很多设计你会觉得莫名其妙。全文围绕几个真实可复现的案例展开日志清洗、用户行为聚合、宽表关联、以及一个从 Pandas 迁移过来的性能改造每个案例我都会把为什么这么写讲透因为 PySpark 最容易出问题的地方不是语法而是你以为它在本地执行其实它在集群上懒执行。我特别想强调一点学 PySpark 最大的误区是把它当分布式的 Pandas来用。语法像执行模型完全不同。Pandas 里一行代码立刻出结果PySpark 里一行代码可能只是往执行计划里加了一个节点真正的计算要等到触发动作才发生。不理解这一点你写的代码能跑但慢得离谱或者报出让你完全摸不着头脑的错。下面的内容我会反复回到这个核心差异上。2. 环境搭建与第一段能跑通的代码2.1 本地模式的选型与安装取舍初学阶段我的建议非常明确先用本地模式local mode不要一上来就折腾集群。原因很实际——本地模式用单机多线程模拟分布式执行你可以在笔记本上完整走通代码逻辑、验证转换结果调试成本极低等逻辑稳定了再往集群上搬这时候出的问题基本都是资源和配置问题而不是业务逻辑问题排查范围小得多。安装上有个关键前提PySpark 依赖 JVM 运行所以本机必须先有 Java 环境。我踩过的第一个坑就在这里——装了最新版 JDK 结果 Spark 起不来报一些看不懂的反射错误。经验是版本要匹配Spark 3.x 系列配 JDK 8 或 JDK 11 比较稳JDK 17 及以上在部分版本上会有兼容问题。至于 PySpark 本身直接 pip 装就行pip install pyspark装完验证一下版本对不对得上import pyspark print(pyspark.__version__)这里补充一个取舍很多人会纠结要不要装完整的大数据全家桶Hadoop、Hive 那一套。我的看法是学习阶段完全没必要。PySpark 的 pip 包自带 Spark 运行时本地读写本地文件、跑 DataFrame 转换足够用了。等你需要读写 HDFS 或者接 Hive 元数据的时候再补那时候你对 Spark 的理解已经足够支撑你判断该装什么。还有一个环境变量的事值得提一句。如果你机器上有多个 Java 版本最好显式指定JAVA_HOME避免 Spark 抓到错误的那个。另外在 Windows 上跑本地模式曾经需要额外配置 Hadoop 的 winutils 工具比较新的 Spark 版本已经改善了很多但如果遇到文件写入相关的奇怪报错先往这个方向查。2.2 一个 SparkSession 该配置哪些参数所有 PySpark 程序的入口都是SparkSession。我见过不少人的代码里到处SparkSession.builder.getOrCreate()这是可以的但真正影响体验的是那几个配置项。下面这个是我本地开发时常用的模板from pyspark.sql import SparkSession spark ( SparkSession.builder .appName(learning-pyspark) .master(local[*]) .config(spark.driver.memory, 4g) .config(spark.sql.shuffle.partitions, 8) .config(spark.sql.adaptive.enabled, true) .getOrCreate() )逐个说清楚为什么这么配。master(local[*])里的星号表示用本机所有可用核心写死成local[4]就是限定四线程调试并行相关的 bug 时限定线程数很有用。spark.driver.memory控制 driver 端内存本地模式下 driver 同时承担执行职责给太小会在聚合阶段 OOM。spark.sql.shuffle.partitions这个参数极其重要默认值是 200。本地跑小数据时 200 个分区意味着每次 shuffle 要拆出 200 份任务绝大部分是空跑调度开销远超计算本身你会觉得这么点数据怎么这么慢。我本地一律设成 8 或 CPU 核数的两三倍。反过来在集群上跑大数据200 又可能太小导致单分区数据量过大这个后面性能那节会展开。spark.sql.adaptive.enabled是自适应查询执行开启后 Spark 会根据运行时统计信息自动合并小分区、调整 join 策略。这个开关在新版本默认开启显式写出来是为了让读代码的人知道它存在。注意本地模式下的性能表现和集群模式差异很大本地调优的经验不能直接照搬到集群。本地模式的价值在于验证逻辑正确性不要用它去推断集群上的性能瓶颈。2.3 数据读取从 CSV 到 DataFrame 的第一道关读取是最容易出问题的环节因为数据脏。以一个常见的日志 CSV 为例我习惯把 schema 显式声明出来而不是让它自己推断from pyspark.sql.types import StructType, StructField, StringType, LongType, TimestampType schema StructType([ StructField(user_id, StringType(), True), StructField(event_time, TimestampType(), True), StructField(event_type, StringType(), True), StructField(duration_ms, LongType(), True), ]) df ( spark.read .option(header, true) .option(inferSchema, false) .schema(schema) .csv(data/events/*.csv) )为什么不用inferSchematrue因为它会额外扫一遍全量数据来推断类型数据量大时这一趟扫描的成本非常可观而且推断结果经常不合预期比如纯数字的 ID 被推断成整型前导零全丢了。显式声明 schema 是一次性的成本换来的是读取快、类型可控、出错早。字段的第三个参数nullable我统统写True因为真实数据里空值太常见了先把容错留出来清洗逻辑单独处理。路径里的*.csv是通配符Spark 天然支持按目录批量读这一点比 Pandas 舒服很多。读完之后第一件事永远是看数据长什么样df.printSchema() df.show(5, truncateFalse) df.count()show默认只显示 20 行且会截断长字符串truncateFalse关掉截断。count()会触发一次全量扫描数据大时慎用但初期确认数据完整性时值得。这里就体现出懒执行的特点了前面 read 那一串其实没读数据直到show或count才真正触发。3. DataFrame 转换的核心逻辑与高频操作3.1 懒执行与执行计划理解这一点少走一半弯路我把它放在最前面讲因为它是理解一切性能问题的钥匙。PySpark 的转换操作分两类transformations转换和actions动作。select、filter、withColumn、join、groupBy这些都是转换它们不会立刻执行只是往逻辑执行计划里追加一个节点show、count、collect、write才是动作触发真正的计算。这个设计带来的直接好处是优化空间。因为 Spark 看到了你的完整意图它可以做一些你没想到的优化比如你写了select三个字段后又filterSpark 会把 filter 下推到读取阶段只读需要的列和行这叫谓词下推和列裁剪比如两个连续的filter会被合并。这些优化在 Pandas 里需要你手动安排顺序。坏处也很明显出错位置和代码位置对不上。你可能在第 20 行写了个类型不匹配的表达式报错却发生在第 80 行的show上堆栈信息还特别长。我的应对习惯是每写几个转换就插一个df.show(1)或者df.explain()看看explain()会打印物理执行计划能看到有没有发生 shuffle、用的什么 join 策略调试性能问题时这是第一个该用的工具。还有一个必须知道的点重复计算。因为懒执行如果你对同一个 DataFrame 连续做多个动作Spark 默认会从头重算整条链路。比如你先count()再show()数据会被读两遍、算两遍。解决办法是df.cache()把中间结果落到内存这个后面会细说。3.2 列操作的三种写法与该选哪个PySpark 操作列有三套 API字符串表达式、pyspark.sql.functions里的函数、以及Column对象的方法。新手容易混着用导致报错我建议以 functions 为主理解清楚另外两种什么时候用。最简单的是字符串表达式适合简单场景df.select(user_id, duration_ms) df.filter(duration_ms 1000) df.groupBy(event_type).count()方便但表达力有限做不了复杂计算。真正的活儿靠functionsfrom pyspark.sql import functions as F result ( df .withColumn(duration_sec, F.col(duration_ms) / 1000.0) .withColumn(hour, F.hour(event_time)) .filter(F.col(event_type).isin(click, view)) .select(user_id, hour, duration_sec) )F.col(x)拿到列引用然后可以用/、、这些运算符也可以用F.when().otherwise()做条件判断。注意一个容易踩的点如果列名和 Python 内置名或者 DataFrame 方法名冲突必须用F.col显式引用。我遇到过表里有个字段叫count直接写df.select(count)在某些版本上会出问题改成F.col(count)就稳了。第三种是链式方法像df.filter(df.duration_ms 1000)写起来像 Pandas但可读性和组合性都不如 functions 风格复杂逻辑一长就乱。我的实践是简单选择用字符串所有计算逻辑统一用 F 系列代码风格一致别人接手也快。3.3 空值、类型与字符串清洗的实战套路真实数据里空值处理占清洗工作量的一大半。PySpark 提供na系列方法但它的行为和 Pandas 有微妙差异我列几个常用场景# 删除关键字段为空的行 df_clean df.na.drop(subset[user_id, event_time]) # 特定列填默认值 df_filled df.na.fill({duration_ms: 0, event_type: unknown}) # 用条件表达式做精细替换 df_precise df.withColumn( duration_ms, F.when(F.col(duration_ms).isNull() | (F.col(duration_ms) 0), 0) .otherwise(F.col(duration_ms)) )na.drop的subset参数很关键——不带它会把任何字段为空的行都删掉经常误伤。第三个例子里我用when同时处理了空值和负值这在时长、金额类字段上很常用因为负数通常是采集错误。类型转换用cast但转换失败不会报错而是返回 null这是个隐蔽的坑df.withColumn(duration_ms, F.col(duration_ms).cast(long))如果原字段里混了abc这种值cast 之后变 null你后面统计才发现数据莫名少了一批。我的习惯是转换前后各做一次df.filter(F.col(x).isNull()).count()对比差值就是转换失败的条数值得查清原因。字符串清洗高频操作是去空格、大小写统一、正则提取。F.trim去首尾空格F.lower/F.upper统一大小写F.regexp_extract(col, pattern, group)提取F.regexp_replace替换。这些组合起来能处理掉绝大部分脏数据。经验是先做 distinct 看有哪些异常值再写规则比拍脑袋写一堆 if-else 靠谱得多。用df.select(event_type).distinct().show(50)扫一眼往往能发现Click、click 、CLICK这种同义不同形的问题。4. 聚合与关联把业务问题翻译成算子4.1 groupBy 聚合的正确姿势与常见误解分组聚合是分析场景的主力。基本形态很直观from pyspark.sql import functions as F agg_df ( df_clean .groupBy(user_id) .agg( F.count(*).alias(event_cnt), F.sum(duration_ms).alias(total_duration), F.avg(duration_ms).alias(avg_duration), F.countDistinct(event_type).alias(distinct_types), F.max(event_time).alias(last_active), ) )这里有个极易踩的坑count(*)和count(col)的行为不一样。前者统计行数后者只统计该列非空的行数。做用户行为计数时如果用count(event_type)而 event_type 恰好有空值你的总数就偏少了还很难发现。countDistinct在数据量大时很贵因为要去重。如果只需要近似去重比如估算 UV用F.approx_count_distinct会快非常多误差在可接受范围内。这个取舍在实际业务里很有价值比如千万级用户的行为分析精确去重可能要跑十几分钟近似算法几十秒出结果误差 1% 以内很多场景完全够用。另一个高频需求是分组后取 TopN。不能简单groupBy然后orderBy因为那样是全局排序。正确做法是窗口函数from pyspark.sql.window import Window w Window.partitionBy(user_id).orderBy(F.col(duration_ms).desc()) top_df ( df_clean .withColumn(rn, F.row_number().over(w)) .filter(F.col(rn) 3) .drop(rn) )partitionBy定义分组边界orderBy定义组内排序row_number给组内每行编号筛前三个就是每组 Top3。如果用rank()并列的会得相同编号并跳过后续编号dense_rank()则不会跳号三者语义不同看业务需要选。4.2 join 的类型选择与 shuffle 代价多表关联在 Spark 里比 Pandas 更需要谨慎因为 join 通常意味着 shuffle——把相同 key 的数据拉到同一个分区。这个网络传输和磁盘落地的开销是大头。user_wide ( events_df.join(profile_df, onuser_id, howleft) .join(F.broadcast(device_df), ondevice_id, howleft) )how支持inner、left、right、full、left_semi、left_anti。后两个特别有用left_semi只保留左表能匹配上的行但不增加列相当于WHERE EXISTSleft_anti反过来只保留匹配不上的相当于WHERE NOT EXISTS。做数据质量检查时我常用left_anti找出事件表里有但用户表里没有的用户这类孤儿数据是很多统计口径对不上的根源。关于broadcast这是最值得掌握的优化手段之一。当右表足够小默认阈值 10MB可配可以用F.broadcast()提示 Spark 把它广播到每个执行节点这样 join 就变成每个分区各自本地关联彻底避免了大表的 shuffle。我处理事件表关联设备维表这种场景时几乎必用。但反过来如果把大表错误地标记成 broadcastdriver 端会先把这个表收集起来直接 OOM 或者卡死所以这个提示只能给小表。还有一个隐蔽问题join 键的数据类型必须一致。左表 user_id 是 string右表是 longSpark 不会自动转结果是全部匹配不上而且不报错。我遇到过一次两表各几十万行join 完是零行查了半小时才发现是类型问题。现在我的习惯是 join 前先printSchema对一眼。4.3 union、pivot 与 explode处理结构变形的三件套除了聚合和关联还有几类结构变换用得特别多。union合并结构相同的 DataFrame注意它是按位置对齐字段而不是按名字两张表字段顺序不一致会静默错位这是很危险的行为。稳妥做法是先select到相同顺序再 unioncols [user_id, event_type, duration_ms] combined df_a.select(cols).unionByName(df_b.select(cols))unionByName按名字对齐更安全我基本都用这个版本。pivot把行转列做报表时常用pivot_df ( df_clean .groupBy(user_id) .pivot(event_type, [click, view, share]) .agg(F.count(*)) )务必给 pivot 指定第二个参数取值列表否则 Spark 要额外扫一遍拿 distinct 值数据大时这一趟能翻倍耗时。而且不指定的话每次运行生成的列可能不一样下游逻辑就不稳定了。explode把数组字段拆成多行处理 JSON 类数据必备df_exploded df_clean.select( user_id, F.explode(tags).alias(tag) )反向操作是collect_list/collect_set聚合成数组。这里有个坑collect_list 不保证顺序如果你需要按时间顺序聚合得先用sort_array或者配合窗口函数处理不能指望聚合过程保序。5. 性能优化从能跑到跑得快的几个关口5.1 分区数调优的计算思路分区数是 PySpark 性能调优的第一变量。太多了任务调度开销压倒计算太少了单分区数据量过大导致 OOM 和长尾。调的时候有个经验公式可以参考目标单分区数据量控制在 100MB 到 200MB 之间。假设你要处理 20GB 数据那大概需要 100 到 200 个分区。这个数字来自工程实践中的经验值不是硬性标准但作为起点很实用。三个关键参数需要区分开参数作用阶段本地开发建议集群建议spark.sql.files.maxPartitionBytes读文件时切分保持默认 128MB保持默认spark.sql.shuffle.partitionsshuffle 后分区数设 8 到 CPU 核数按数据量算数百到数千spark.default.parallelismRDD 操作默认并行度较少用通常设为核数的 2 到 3 倍shuffle.partitions是最常调的一个。除了全局设置还可以在代码里动态改spark.conf.set(spark.sql.shuffle.partitions, 64)或者对特定 DataFrame 用repartition/coalesce。这里有个重要区别repartition会触发全量 shuffle可以增也可以减分区数coalesce只做合并、不 shuffle只能减少分区。写文件前想减少文件数量用coalesce(1)很常见但要小心——所有数据会被挤到一个分区处理几 GB 数据可能要跑很久甚至 OOM所以它适合输出一个汇总小文件不适合大数据导出。5.2 cache 与 persist什么时候该用什么时候是负担前面提到重复计算的问题。假设有这样一段逻辑base df.filter(...).withColumn(...) # 复杂的清洗逻辑 print(base.count()) # 触发第一次完整计算 result base.groupBy(...).agg(...) # 又要从头算一遍清洗 result.show()清洗逻辑跑了两遍。解决办法base df.filter(...).withColumn(...).cache()cache()之后第一次触发时会把结果存到内存后续复用直接读缓存。但不是所有场景都该 cache这里有几个判断依据这个 DataFrame 会被多次使用吗只用一次就 cache纯属浪费内存。缓存的数据量放得下吗如果数据比可用内存大得多缓存会退化成溢出到磁盘反而更慢。缓存真的被用到过吗cache 是懒的你不触发动作它就不会缓存很多人写了 cache 但后面没复用白白占着内存没释放。用完记得df.unpersist()尤其在长时间运行的会话里。我在本地调试时经常忘了这茬跑几个案例之后内存越来越紧最后 driver OOM重启才发现是前面的缓存没清。persist比cache多一个级别参数可以选择只在磁盘或者内存磁盘。默认的cache等价于MEMORY_AND_DISK对大多数场景够用。真正需要精细控制存储级别的情况不多。5.3 从 Pandas 迁移过来最容易犯的三个错我见过太多把 Pandas 代码逐行翻译成 PySpark的案例能跑但慢得多。三个典型问题第一个是滥用 UDF。Pandas 里习惯了apply逐行处理转到 PySpark 就写成F.udf。问题是 Python UDF 要在 JVM 和 Python 进程之间序列化传输数据性能比内置函数慢一个数量级。能用F.when、F.regexp_extract、F.coalesce这些内置函数表达的一律不要用 UDF。如果确实需要自定义逻辑优先考虑pandas_udf向量化走 Arrow 传输快得多最后才选普通 UDF。第二个是不理解 shuffle。Pandas 的merge和groupby在内存里做PySpark 对应操作要 shuffle。如果代码里连着三四个groupBy和join中间没有任何过滤那 shuffle 会一次接一次数据量一直没降下来。我的做法是尽早过滤和裁剪把filter和select往前放让进入 shuffle 的数据尽可能小。第三个是把结果 collect 回本地。collect()会把所有分区的数据拉到 driver 内存数据稍大直接 OOM。这是最经典的事故。正确做法是聚合或截断后取# 危险数据大时直接崩 all_rows df.collect() # 安全先聚合结果必然很小 summary df.groupBy(dt).count().collect() # 或者只要样本 sample_rows df.limit(100).collect()实操心得调试期间想看数据用show(n)而不是collect()。show本身也会把 n 行拉到 driver但 n 有上限风险可控。养成这个习惯能省掉很多次重启。5.4 数据倾斜最顽固的性能杀手如果一段作业卡在最后几个任务上迟迟不结束而前面几百个任务早就完成了八成是数据倾斜——某个 key 的数据量远超其他 key导致处理它的那个分区成了长尾。典型场景按 user_id 聚合时恰好有几个超级用户贡献了大部分数据或者关联时某个维度值特别集中。识别方法很简单df.groupBy(user_id).count().orderBy(F.col(count).desc()).show(20)如果第一名是第二名的几十倍那就是倾斜了。处理手段有几种按成本从低到高。一是过滤掉异常 key 单独处理比如那几个明显不合理的超级 ID单独算完再合并结果。二是加盐打散给倾斜 key 拼上随机后缀先局部聚合再把后缀去掉做第二次聚合这套流程稍复杂但很通用salt_df ( df.filter(F.col(user_id) skewed_id) .withColumn(salt, (F.rand() * 10).cast(int)) .groupBy(user_id, salt) .agg(F.sum(amount).alias(partial_sum)) .groupBy(user_id) .agg(F.sum(partial_sum).alias(total)) )三是开启自适应执行新版本 Spark 的 AQE 能自动检测倾斜的分区并拆分很多时候开了就能缓解代码不用改。这个开关是spark.sql.adaptive.skewJoin.enabled配合spark.sql.adaptive.enabled一起用。四是如果是 join 倾斜可以尝试把大表拆小、小表广播把 shuffle join 转成 map join。这个思路在很多维表关联场景下效果极好。6. 实战案例一条完整的日志清洗与用户分析链路6.1 案例背景与数据处理目标这一段我把前面的知识点串成一个能复现的完整案例。场景是常见的用户行为日志分析原始数据是多个 CSV 文件包含用户 ID、事件时间、事件类型、时长字段存在空值、大小写混乱、时间格式不统一、重复记录等问题。目标是产出一张用户维度的宽表包含行为次数、各类事件分布、活跃时间跨度、Top 事件等供后续分析使用。数据规模假设是千万级这个量级单机 Pandas 会很吃力正好用来体会 PySpark 的价值。虽然我这里用本地模式演示但代码结构是照着集群跑的思路写的直接搬到集群上改一下master和输出路径就能用。整个链路我分成四步读取与规范化、清洗与去重、聚合与关联、输出与验证。每一步都会明确说明做了什么、为什么这么做、有什么坑。6.2 读取与规范化把脏数据挡在第一道关口读取阶段我坚持显式 schema理由前面说过。但日志数据常常不是理想格式时间字段尤其麻烦from pyspark.sql import functions as F from pyspark.sql.types import StructType, StructField, StringType, LongType raw_schema StructType([ StructField(user_id, StringType(), True), StructField(event_time, StringType(), True), StructField(event_type, StringType(), True), StructField(duration_ms, StringType(), True), ]) raw ( spark.read .option(header, true) .option(nullValue, ) .schema(raw_schema) .csv(data/raw_events/*.csv) )注意这里时间、时长我都先读成 StringType。原因是源数据格式不统一——有的时间是2024-01-05 10:23:11有的是2024/01/05 10:23:11有的是 unix 时间戳。如果直接读成 TimestampType格式不匹配的直接变 null数据静默丢失。先读成字符串再显式转换能让你看到转换失败的情况这是我处理脏数据的核心原则之一。规范化阶段把格式统一normalized ( raw .withColumn(event_time_fixed, F.regexp_replace(event_time, /, -)) .withColumn(event_time_ts, F.to_timestamp(event_time_fixed, yyyy-MM-dd HH:mm:ss)) .withColumn(duration_ms_num, F.col(duration_ms).cast(long)) .withColumn(event_type_clean, F.lower(F.trim(event_type))) )to_timestamp指定格式串比让它自动猜可靠得多。regexp_replace把斜杠统一成横杠这样格式就齐了。这里用到的都是内置函数没有 UDF性能有保障。转换完必须核对损失total normalized.count() bad_time normalized.filter(F.col(event_time_ts).isNull()).count() bad_duration normalized.filter(F.col(duration_ms_num).isNull()).count() print(f总行数 {total}, 时间转换失败 {bad_time}, 时长转换失败 {bad_duration})这个对账动作我强烈建议每个清洗流程都做一次。数字出来之后如果失败率明显偏高说明格式假设错了得回去看原始数据别在这一步硬扛下去。6.3 清洗去重规则要基于观察而非猜测清洗规则不能凭空写。我的第一步永远是把候选字段的取值分布看一眼normalized.groupBy(event_type_clean).count().orderBy(F.col(count).desc()).show(30)看完之后通常会发现几类问题同一含义多种写法click和clk、空值、异常高频值可能是埋点 bug 或者测试数据。根据观察结果写映射cleaned ( normalized .filter(F.col(user_id).isNotNull()) .filter(F.col(event_time_ts).isNotNull()) .withColumn( event_type_final, F.when(F.col(event_type_clean).isin(click, clk), click) .when(F.col(event_type_clean).isin(view, pv), view) .otherwise(other) ) .withColumn( duration_ms_final, F.when(F.col(duration_ms_num) 0, 0) .otherwise(F.col(duration_ms_num).cast(long)) ) )去重是另一个关键步骤日志重复上报很常见。判断重复要基于业务定义我这里用用户时间事件类型三字段deduped cleaned.dropDuplicates([user_id, event_time_ts, event_type_final])dropDuplicates带子集参数只按指定字段判重。注意它的行为如果只给子集Spark 会从重复行里保留任意一条不保证保留哪条。如果你的字段之间有其他差异需要保留特定那条得先用窗口函数排序再筛。不带参数的话按全字段判重更保守但可能漏掉真正的重复比如两个字段有小差异。这里对账一下去重效果print(f清洗后 {cleaned.count()}, 去重后 {deduped.count()})去掉了多少、比例是否合理心里要有个数。去掉 30% 以上通常说明判重规则太激进得回头查。6.4 聚合与宽表构建把指标算准聚合这一步的要点是指标定义清晰。我按用户维度算几类指标from pyspark.sql import Window w_user Window.partitionBy(user_id) user_agg ( deduped .groupBy(user_id) .agg( F.count(*).alias(total_events), F.sum(F.when(F.col(event_type_final) click, 1).otherwise(0)).alias(click_cnt), F.sum(F.when(F.col(event_type_final) view, 1).otherwise(0)).alias(view_cnt), F.sum(duration_ms_final).alias(total_duration), F.min(event_time_ts).alias(first_active), F.max(event_time_ts).alias(last_active), F.countDistinct(event_type_final).alias(distinct_events), ) )用F.sum(when(...))算条件计数是 Spark 里最常用的手法之一比先 filter 再 join 高效得多——一次 groupBy 把所有指标都算完只 shuffle 一次。如果写成click 事件 filter 出来 groupBy 一次view 事件再 groupBy 一次然后 join那就是三次 shuffle性能差好几倍。日期差计算活跃跨度user_agg2 user_agg.withColumn( active_days, F.datediff(F.col(last_active), F.col(first_active)) )datediff算两个时间戳相差的天数。这里有个细节它只比较日期部分不计时间如果需要精确到小时得用unix_timestamp相减再除。这个差异在做留存、时长类指标时要搞清楚否则口径会有偏差。再关联一张用户基础信息表得到完整宽表profile_schema StructType([ StructField(user_id, StringType(), True), StructField(register_date, StringType(), True), StructField(channel, StringType(), True), ]) profile ( spark.read .option(header, true) .schema(profile_schema) .csv(data/user_profile/*.csv) .withColumn(register_date, F.to_date(register_date, yyyy-MM-dd)) .na.drop(subset[user_id]) ) wide_table user_agg2.join(profile, onuser_id, howleft)关联用left保证行为数据不丢。这里可以用F.broadcast(profile)优化因为用户信息表通常比行为表小得多。能不能用 broadcast 的判断标准是表大小——几十 MB 以内的维表广播收益明显。关联完要检查匹配率这是数据质量的常规体检matched wide_table.filter(F.col(channel).isNotNull()).count() print(f行为用户 {user_agg2.count()}, 匹配到维表 {matched})匹配率过低说明两张表的 ID 体系可能有问题比如大小写、前后空格、编码不一致。我遇到过维表 ID 带空格的join 全不匹配trim一下就好了。6.5 输出与结果校验别在最后一步翻车输出看似简单但有几个坑。第一个是文件数量repartition出来的分区数决定输出文件数几百个分区就是几百个小文件下游用起来很烦( wide_table .repartition(8) .write .mode(overwrite) .option(header, true) .parquet(output/user_wide) )我这里用repartition(8)是因为结果表不会太大8 个文件比较合适。如果结果数据量很大就按前面说的 100MB 到 200MB 每文件去算分区数。不要无脑coalesce(1)前面说过那是把所有数据挤到一个分区。格式上我优先选 parquet 而不是 CSV理由有三个列式存储读取快、自带 schema 不用每次推断、压缩率高。如果下游非要用 CSV再单独导一份。第二个坑是mode参数。overwrite覆盖、append追加、error默认目标存在就报错、ignore。默认的error经常让人误以为没写成功其实是目标路径已存在。我一般显式写overwrite或者append语义清楚。输出后回读验证是必做的check spark.read.parquet(output/user_wide) print(check.count()) check.printSchema() check.show(5)顺便做几个业务校验比如总事件数是否和清洗阶段对得上、有没有关键字段全为空、分区文件有没有 0 字节的。这些检查花不了几分钟能提前发现很多问题。没有校验的产出等于没产出这是我这些年做数据流程最深的体会之一。7. 常见问题排查与避坑清单7.1 报错类型速查与定位思路PySpark 的报错信息长、层级深很多人第一眼就懵了。我的经验是只看最后两段和第一个Caused by中间那一大堆是调用栈对你定位问题帮助不大。报错特征大概率原因排查动作OutOfMemoryError: Java heap spacedriver 或 executor 内存不足检查是否collect()了大结果或 reduceByKey 前没预聚合AnalysisException: cannot resolve字段名写错、大小写、前后空格printSchema()对照注意字段名是否有空格AnalysisException: resolved attribute(s) ... mismatch自关联或多次 join 后列引用歧义给 DataFrame 起别名用df.alias(a)[col]明确引用Py4JJavaError 长堆栈多种原因重点看 Caused by看 Caused by 里的具体描述通常是类型或空值问题任务卡在 99% 不结束数据倾斜或长尾任务查 key 分布考虑 AQE 或加盐FileNotFoundException路径不对、通配符没匹配到文件先hadoop fs -ls或本地ls确认路径转换后大量 nullcast 失败被静默转 null转换前后对比 null 数量有几个问题值得单独说。自关联的列歧义是老问题比如df.join(df, ...)之后引用同名列Spark 不知道你说哪个报歧义错误。解决办法是先起别名a df.alias(a) b df.alias(b) joined a.join(b, F.col(a.user_id) F.col(b.ref_user_id), inner)字段名带空格也是隐形杀手F.col(user id)看着没问题但很多场景会出岔子能提前 rename 就 rename。7.2 开发期的高效调试习惯调试 PySpark 有几个习惯能显著提升效率都是被坑出来的。第一小数据集验证逻辑。我会从真实数据里抽 1000 行存成一个小文件所有逻辑先在这个小文件上跑通。本地跑 1000 行是秒级的跑全量可能要几十分钟迭代效率差几十倍。逻辑确认无误再上全量这是最基本的工程习惯。第二explain()看执行计划。前面提过df.explain(True)会打印详细的逻辑和物理执行计划。看到Exchange节点就知道发生了 shuffle看到BroadcastHashJoin说明广播生效了看到SortMergeJoin说明是普通 shuffle join。这些信息对你判断性能很有用。第三show加截断控制。show(20, truncate50)把长字符串截断输出干净很多。verticalTrue可以让每条记录竖着显示字段多的时候特别有用df.show(3, verticalTrue)第四分区数据量分布检查。数据倾斜的早期发现很简单df.withColumn(pid, F.spark_partition_id()).groupBy(pid).count().orderBy(count).show()看最大分区和最小分区的比值差得太多就要警惕了。这个比等作业跑一半卡住再查要主动得多。第五本地会话频繁创建开销大。SparkSession创建一次要几秒到十几秒反复创建会很烦。在交互式环境里比如 notebook保持一个会话复用只在改配置时重启。7.3 几个我踩过的真实坑与规避方法有几个坑记忆深刻都是文档里不太会写但实际很痛的。坑一count()之后的show()数据不一致。曾经遇到count()返回 100 万show()出来的数据却是旧版本。原因是中间某个 DataFrame 依赖了带随机性的转换比如F.rand每次计算重新生成。带随机或非确定性函数的链路多次 action 结果可能不同。解决方法是加cache把结果固定住。坑二orderBy后limit依然扫全量。我以为orderBy(...).limit(10)会只算前十个其实还是全量排序再取前十个。真正想加速得用df.orderBy(...).limit(10)配合分区裁剪或者对数据先过滤缩小范围。这个属于 SQL 引擎的常见特性但对新手确实反直觉。坑三空 DataFrame 的show什么都没输出。第一次遇到以为程序挂了其实是真的没数据。原因往往在路径通配符没匹配到文件或者上游 filter 全过滤掉了。排查时先count()确认是 0 还是程序卡住。坑四默认分区数在小数据上极其浪费。前面提过shuffle.partitions默认 200本地小数据每个任务可能就几行调度开销占绝大部分。我现在的模板里一律设小这个改动带来的提速是肉眼可见的。坑五cache了忘记unpersist。交互式环境里连着跑几个案例内存慢慢被缓存占满最后报 OOM莫名其妙。现在我的习惯是写完一段分析就spark.catalog.clearCache()清一下或者在每段结尾显式 unpersist。最后再分享一个小技巧给SparkSession的appName起一个有辨识度的名字然后开 Spark UI 看作业详情。本地模式启动时日志会打印 UI 地址进去能看到每个 stage 的耗时、shuffle 读写量、分区大小分布。这些信息比日志有用得多尤其是定位慢在哪一步的时候。我早期排查一个跑了四十分钟的作业就是通过 UI 发现 95% 时间耗在一次不必要的join上去掉之后降到三分钟。7.4 从学习到落地的路径建议如果你现在准备把 PySpark 用起来我建议按这个顺序推进别贪快。先花半天把本地环境跑通写几个最小的 DataFrame 操作select、filter、groupBy各来一遍体会懒执行和 action 的区别。然后找一个自己熟悉的 Pandas 任务用同样的数据在 PySpark 里重写一遍对比结果是否一致——这个对比能暴露很多理解偏差比如空值处理、类型转换、排序稳定性。第三步才是引入大数据量观察分区和性能表现把前面讲的调优手段逐个试一遍看效果。真正上手项目时我的建议是先保证正确再谈优化。很多人一上来就想着 repartition、broadcast、cache结果逻辑还没跑对调优成了盲调。数据量不大的阶段默认配置往往就够用等确实出现慢的时候再按看执行计划、查分区分布、定位 shuffle 和倾斜这个顺序下手效率高得多。还有一点得说清楚PySpark 不是万能的。几百万行以下的数据单机 Pandas 可能反而更快因为省掉了分布式调度和序列化的开销。我现在的判断标准大概是几百万行以内用 Pandas上了千万行或者单机内存明显吃紧才上 Spark。工具是拿来解决问题的不是拿来炫技的选对场景比会用工具本身更重要。这个判断力是在反复折腾不同规模的作业之后慢慢养成的急不来。