1. 为什么 DataFrame 算子值得单独拎出来讲
写过一段时间 Spark 的人都经历过这个阶段：RDD 用得很顺手，map、filter、reduceByKey闭着眼睛都能写，突然转到 DataFrame，发现 API 长得完全不一样，select、withColumn、groupBy、agg一大堆，写起来像是换了一门语言。更让人抓狂的是，明明逻辑一样，别人写的 DataFrame 代码跑得比你快好几倍，你盯着执行计划看半天也看不出问题在哪。
这就是我想把 DataFrame 常用算子 API 单独拉出来聊一遍的原因。DataFrame 不是 RDD 的语法糖，它背后挂着一整套 Catalyst 优化器和 Tungsten 执行引擎，你写的每一个算子最终会不会被优化、会不会触发 shuffle、会不会把数据拉到 Driver 端，结果差别巨大。同样一个聚合需求，用groupBy().agg()和用collect_list()之后再处理，数据量上到千万级就是几分钟和几小时的差距。
这篇内容适合三类人看：刚接触 Spark SQL 不久、能写 SQL 但不太会写 DataFrame API 的同学；已经能跑通任务、但经常被数据倾斜和 OOM 折磨的中级使用者；以及想搞清楚"为什么这么写更快"、准备做性能调优的从业者。我会按照实际写代码的顺序，从列操作、行过滤、聚合、连接、窗口一路讲下去，每个算子都配上能直接抄的参数写法和我自己踩过的坑。
需要提前说明一点：下面涉及的所有代码示例都基于 Spark 3.x 的 PySpark 和 Scala API，两者在算子命名上高度一致，个别差异我会单独标注。示例里的表名、字段名都是虚构的业务场景，你替换成自己的即可。文中提到的参数取值、分区数计算、内存配置，都是基于常见生产实践的合理推断，具体到你的集群还需要实测校准。
2. 先把 DataFrame 算子的分类逻辑理清楚
2.1 按是否触发 shuffle 来分类，比按字母排序有用得多
官方文档把算子按字母顺序排了一遍，查起来方便，但学习的时候特别容易迷路。我自己的习惯是按"是否触发 shuffle"来分，因为 shuffle 是 Spark 里最贵的操作，一次 shuffle 意味着磁盘 IO、网络传输、序列化反序列化三重大山。你写代码的时候脑子里有这个分类，就能预判这段逻辑大概要跑多久。
窄依赖类算子：select、filter、withColumn、drop、distinct（严格说 distinct 会 shuffle，但dropDuplicates在部分场景可以避免）、cast、when/otherwise。这类算子只作用在单个分区内的每一行上，不会移动数据，代价极低，你可以放心大胆地连着写十几个。
宽依赖类算子：groupBy().agg()、join、orderBy、repartition、dropDuplicates（全局去重时）、window函数里的partitionBy。这类算子会触发数据在网络间重新分布，写的时候要格外小心，能合并的尽量合并，能提前过滤的坚决提前过滤。
我见过太多这样的代码：先join两张千万级的表，再filter掉 90% 的数据。正确顺序显而易见，但写的时候人就是容易顺着业务逻辑一路往下写。所以我的建议是，写完一段 DataFrame 逻辑，回头看一眼：过滤条件能不能往前挪？聚合能不能先在子查询里做一轮预聚合？
2.2 惰性求值和 Action 算子的边界
DataFrame 的算子是惰性的，select、filter、join这些叫 Transformation，写完不会立刻执行，只是往逻辑计划树上挂节点。真正触发计算的是 Action 算子：show、count、collect、take、write、foreach。
这个机制带来的第一个实际影响是：调试的时候如果你想看中间结果，必须加一个 Action。很多人写了一大串 Transformation，然后用print(df)想看数据，结果只打印出 schema，以为代码有问题。其实加个df.show(5)就好了。
第二个影响更关键：同一个 DataFrame 被多次 Action 触发时会重复计算。比如你先df.count()看数据量，再df.groupBy(...).count().show()，如果没有做 cache，底层的扫描会执行两遍。生产环境里我一般会在复用超过两次的 DataFrame 上打一个cache()或persist(StorageLevel.MEMORY_AND_DISK)，用完立刻unpersist()。这里的取舍是：cache 会占内存，如果数据量超过可用内存，反而会因为频繁换入换出拖慢速度，所以MEMORY_AND_DISK通常是更稳妥的默认选择。
还有一个容易被忽略的点：DataFrame 的列名解析在 Analysis 阶段完成，不是运行时。这意味着如果你对列名拼错了，报错会在 Action 触发时以AnalysisException的形式抛出来，而不是在你写select("wrong_col")的那一行。排查这类问题的时候，先看异常栈里的 "cannot resolve" 关键字，基本能定位到具体的列。
3. 列操作与行过滤：日常写得最多的一类算子
3.1 select、withColumn、withColumnRenamed 的取舍
这三个算子是日常工作里出场率最高的。select用于挑选和变换列，withColumn用于新增或覆盖列，withColumnRenamed用于改列名。
from pyspark.sql import functions as F # 挑选列并做简单变换 df.select( "user_id", F.col("amount").cast("double").alias("amount_d"), F.upper(F.col("city")).alias("city_upper") ) # 新增一列计算字段 df.withColumn("amount_with_tax", F.col("amount") * 1.06) # 改列名 df.withColumnRenamed("user_id", "uid")看起来很简单，但有两个坑值得说。
第一个坑：withColumn连续调用会不断膨胀逻辑计划。我见过一段代码连续调了 40 多个withColumn，生成的执行计划里嵌套了几十层 Project 节点，Catalyst 优化起来非常吃力，跑起来也慢。正确做法是能合并的用一次select全部写完，或者用withColumns（Spark 3.3+ 提供，一次传入 dict）。如果你的版本不支持withColumns，可以先把所有表达式攒到一个列表里，最后一次性select展开。
# 不推荐：连续 withColumn df2 = df.withColumn("a", F.col("x") + 1) \ .withColumn("b", F.col("y") * 2) \ .withColumn("c", F.col("z") - 3) # 推荐：一次性 select 展开 exprs = [ F.col("x") + 1, F.col("y") * 2, F.col("z") - 3, ] df2 = df.select("*", *[e.alias(n) for e, n in zip(exprs, ["a", "b", "c"])])第二个坑：列名重复问题。select里如果选出了两个同名同义的列，后续引用会报Reference 'col' is ambiguous。常见于 join 之后直接select("*")，两张表都有id列。解决办法是 join 之后立刻用alias或者显式列出需要的列，别偷懒用*。
3.2 filter 和 where：写法差异与下推优化
filter和where在 DataFrame API 里是同一个东西，完全等价，选哪个看个人习惯。真正需要关心的是**谓词下推（Predicate Pushdown）**会不会生效。
# 两种写法等价 df.filter(F.col("dt") == "2024-01-01") df.where("dt = '2024-01-01'")谓词下推的意思是：如果你的数据源是 Parquet、ORC 这类列式存储，且数据按dt做了分区目录，那么过滤条件会被下推到文件扫描层，直接跳过不满足条件的分区目录，压根不读那些文件。这个优化对性能的影响是数量级的。
但下推不是无条件的。如果你在过滤条件里对分区列套了函数，下推就失效了。比如F.date_format(F.col("dt"), "yyyyMM") == "202401"，优化器认不出这跟分区目录的对应关系，只能全表扫描。我在实际项目里见过有人为了统一格式这么写，结果任务从 8 分钟变成 2 小时，改回直接比较字符串后立刻恢复。
注意：判断分区列类型时，如果分区目录是
dt=2024-01-01这种字符串形式，过滤条件也必须用字符串比较，别写F.to_date(F.col("dt")) == ...，同样会打断下推。
3.3 空值处理：fillna、dropna 和 coalesce 的分工
空值在 Spark 里是null，跟空字符串是两回事。处理空值的算子有三组，用途完全不同，混用会出错。
fillna用指定值替换 null，支持传一个 dict 按列指定不同填充值。dropna直接删除含 null 的行，可以指定how="any"（任一列为 null 就删）或how="all"（全为 null 才删），还能用subset限定只检查某些列。coalesce是函数，返回第一个非 null 的值，适合做列间兜底。
# 按列填充 df.fillna({"amount": 0, "city": "unknown"}) # 只检查关键列，任一为空则删除 df.dropna(how="any", subset=["user_id", "order_id"]) # 列间兜底：优先用新值，没有再退回旧值 df.withColumn("final_amount", F.coalesce("new_amount", "old_amount", F.lit(0)))实操心得：dropna在生产环境要非常谨慎。我曾经处理过一批埋点数据，dropna()一执行，数据量从 2 亿掉到 8000 万，后来排查发现是新增的一个可选字段大面积空缺，how="any"把所有缺失这个字段的行全删了。正确做法是永远显式指定subset，明确告诉自己和后来人"我只在乎哪几列不能空"。
3.4 when/otherwise 与 UDF 的边界选择
条件分支在 DataFrame 里用when(...).otherwise(...)写，可以嵌套，像 if-else 一样。
df.withColumn( "level", F.when(F.col("amount") >= 1000, "high") .when(F.col("amount") >= 100, "mid") .otherwise("low") )这里有个非常值得强调的原则：能用内置函数解决的，绝不用 UDF。UDF 分为 Python UDF 和 Pandas UDF 两种，Python UDF 会让每一行数据在 JVM 和 Python 进程之间来回序列化，性能损失巨大，而且 Catalyst 完全无法优化它。Pandas UDF（pandas_udf）基于 Arrow 批量传输，性能好很多，但仍然不如内置函数。
我做过一个粗略对比：对一个千万行表做字符串拼接加条件判断，纯F.when+F.concat大约 20 秒，Python UDF 版本要 4 分半，差距二十倍以上。所以我的判断标准是：简单逻辑一律内置函数；确实需要复杂业务逻辑（比如调第三方库做地理编码）才考虑 Pandas UDF；Python UDF 除非是遗留代码，否则不推荐。
4. 聚合与分组：最容易出性能问题的地方
4.1 groupBy + agg 的标准写法和参数含义
groupBy配合agg是最典型的聚合写法。agg里可以传多个聚合表达式，每个表达式返回一列。
from pyspark.sql import functions as F result = df.groupBy("city", "category").agg( F.count("*").alias("cnt"), F.sum("amount").alias("total_amount"), F.avg("amount").alias("avg_amount"), F.countDistinct("user_id").alias("uv"), F.max("amount").alias("max_amount"), F.collect_set("product_id").alias("products") )几个参数层面的细节值得抠一下。
count("*")和count("col")的结果可能不一样：前者统计行数，后者只统计该列非 null 的行数。如果你的列有 null 且你想统计行数，一定要用count("*")或count(F.lit(1))。
countDistinct在底层实现上会把去重后的值全部收集到内存里做精确去重，当基数很大（比如统计 UV 上千万）时会成为瓶颈甚至 OOM。生产环境里如果对精度要求不是 100%，可以用approx_count_distinct("user_id", 0.05)，第二个参数是相对误差上限，0.05 表示允许 5% 误差，在这个精度下性能通常能提升一个数量级，内存占用也可控。
collect_set和collect_list会把分组内的值收集成数组。这两个算子是内存杀手，如果某个分组的元素数量特别大（比如一个城市有几百万条订单），单个 executor 直接撑爆。我的经验是：除非能确定分组后的元素数量在几千以内，否则别用这两个，改用先聚合再 join 的方式。
4.2 分组后排序：orderBy 放在哪里差别很大
orderBy或者sort是全局排序，会触发一次全量 shuffle 把数据按排序键重新分布，代价很高。常见的错误是把它写在聚合前面。
groupBy(...).agg(...)之后再orderBy是对的，因为聚合后数据量已经大幅缩小。但如果你在groupBy之前orderBy，那个排序基本白做，因为聚合会打乱顺序。
# 不推荐：先排序再聚合，排序白费 df.orderBy("amount").groupBy("city").agg(F.sum("amount")) # 推荐：先聚合，数据量小了再排序 df.groupBy("city").agg(F.sum("amount").alias("total")).orderBy(F.desc("total"))还有一个细节：如果只需要每个分组的前 N 条，别用orderBy然后limit，这仍然是全局排序。正确做法是用窗口函数row_number()（下一节会讲），只在分区内排序，效率高得多。
4.3 数据倾斜的三个实用解法
聚合类操作最怕数据倾斜：某个分组键特别热，比如"未知城市"这一个 key 占了 40% 的数据，那么负责这个 key 的那个 task 要处理几千万行，其他 task 几秒就跑完了，整个 Stage 卡在这一个 task 上。
第一个解法是两阶段聚合加随机前缀。思路是给热点 key 加上随机后缀先做局部聚合，去掉后缀再做一次全局聚合。
import random # 第一阶段：加随机前缀，做局部聚合 df_pre = df.withColumn( "salt", (F.rand() * 10).cast("int") ).withColumn("salted_key", F.concat_ws("_", F.col("city"), F.col("salt"))) partial = df_pre.groupBy("salted_key").agg(F.sum("amount").alias("partial_sum")) # 第二阶段：去掉盐，做最终聚合 final = partial.withColumn( "city", F.split("salted_key", "_")[0] ).groupBy("city").agg(F.sum("partial_sum").alias("total"))第二种解法是热点 key 单独处理。先统计各个 key 的数据量，找出超过阈值的那几个，把热点数据和非热点数据分开聚合，最后union起来。这个方案代码量大一些，但对倾斜特别严重（单 key 占一半以上）的场景最有效。
第三种是广播小表 join 替代聚合。如果倾斜发生在 join 阶段而不是 groupBy 阶段，且其中一张表足够小（默认阈值spark.sql.autoBroadcastJoinThreshold是 10MB），把 Broadcast Join 打开就能把 shuffle join 变成 map 端 join，彻底消除倾斜。
注意：广播阈值不要盲目调大。我试过把阈值设到 500MB，结果 executor 内存被广播变量占爆，任务反而失败。合理的做法是控制在 100MB 以内，同时确认
spark.sql.adaptive.enabled是开启的。
5. Join 与窗口函数：复杂度最高的两块
5.1 五种 Join 类型的语义差异
DataFrame 支持inner、left/left_outer、right/right_outer、full/full_outer、left_semi、left_anti、cross。日常用得最多的是前四种，但semi和anti这两个特别实用，很多人不知道。
# 只保留左表中有匹配的行，但不带出右表的列 df_a.join(df_b, on="user_id", how="left_semi") # 只保留左表中没有匹配的行 df_a.join(df_b, on="user_id", how="left_anti")left_semi等价于WHERE user_id IN (SELECT ...)，left_anti等价于WHERE user_id NOT IN (...)。它们的好处是不带出右表的任何列，输出宽度只跟左表有关，数据量也天然去重（每行左表数据最多保留一份）。
这里必须提一个经典陷阱：NOT IN遇到 null 的坑在left_anti里不存在。SQL 里的NOT IN如果子查询结果包含 null，整个条件会返回 unknown，导致结果为空。left_anti是专门实现的反连接语义，不会有这个问题。所以做反连接一律用left_anti，别自己写NOT IN。
连接的另一个关键点是列名冲突。如果on参数传的是列名列表，join 后同名列会自动合并成一列；如果传的是表达式，同名列会保留两份，需要自己用alias区分，否则后续引用报 ambiguous。
5.2 Join 的类型选择与广播策略
join 的性能取决于执行策略。Spark 有三种策略：Broadcast Hash Join、Sort Merge Join、Shuffled Hash Join。
Broadcast Hash Join 是最快的，因为它不需要 shuffle，把小表完整复制到每个 executor 上，在 map 阶段就完成连接。触发条件是小表体积小于spark.sql.autoBroadcastJoinThreshold。如果你明确知道某张表很小但统计信息不准导致优化器没选广播，可以用F.broadcast()强制广播。
from pyspark.sql.functions import broadcast result = big_df.join(broadcast(small_df), on="user_id", how="left")Sort Merge Join 是大表 join 大表的默认策略，两边都要按 join key 分区排序，再归并。Shuffled Hash Join 在 Spark 3.x 里由 AQE 自动选择，当一侧数据量在分区后变小了就转成哈希表。这些策略的选择逻辑建议不要手动干预，把spark.sql.adaptive.enabled和spark.sql.adaptive.skewJoin.enabled都打开，让 AQE 自己处理。
实操心得：join 之前先做列裁剪和行过滤，比任何调参都有效。我有一个任务，把 join 前select从 60 列减到 8 列、过滤掉历史数据后，运行时间从 40 分钟降到 9 分钟，没改一行业务逻辑。
5.3 窗口函数的核心参数与典型用法
窗口函数是 DataFrame 里最强大也最容易写错的一块。核心结构是Window.partitionBy(...).orderBy(...).rowsBetween(...)。
from pyspark.sql.window import Window w = Window.partitionBy("city").orderBy(F.desc("amount")) # 每个城市金额前三 top3 = df.withColumn("rn", F.row_number().over(w)).filter("rn <= 3") # 累计求和 w_cum = Window.partitionBy("city").orderBy("dt").rowsBetween( Window.unboundedPreceding, Window.currentRow ) df.withColumn("cum_amount", F.sum("amount").over(w_cum))三个参数的含义要分清。partitionBy决定数据怎么分组，同一个分组的数据会落在同一个 task 上。orderBy决定组内排序，只有加了 orderBy，排名类和累计类函数才有意义。rowsBetween决定窗口的边界范围，不写的话默认是从分区开头到当前行（对带 orderBy 的聚合函数）。
row_number()、rank()、dense_rank()的区别必须搞清楚：遇到并列时，row_number依次编号 1、2、3，rank会跳号（1、1、3），dense_rank不跳号（1、1、2）。业务上取"前 N 名"时，如果允许并列并入，用rank或dense_rank；如果要严格限制条数，用row_number。
有个坑我踩过：窗口函数的 partitionBy 键分布不均时同样会倾斜。如果按城市分区，某个城市数据特别多，那个 task 就会特别慢。解法是先在子查询里做过滤或者预聚合，减小单个分区的数据量，或者改用approx类函数降低计算复杂度。
5.4 常用内置函数速查
除了聚合和窗口，日常还有一批函数用得极频繁，我整理成表方便对比查阅。
| 函数 | 用途 | 典型写法 | 注意事项 |
|---|---|---|---|
F.lit | 生成常量列 | F.lit(1) | 和列名同名时要加别名 |
F.concat_ws | 多列拼接 | F.concat_ws("-", "a", "b") | 自动忽略 null，比 concat 安全 |
F.split | 字符串切分 | F.split("s", "_")[0] | 下标从 0 开始，越界返回 null |
F.regexp_extract | 正则提取 | F.regexp_extract("s", "(\\d+)", 1) | 第三个参数是分组序号，0 表示整体 |
F.date_format | 日期格式化 | F.date_format("dt", "yyyy-MM") | 会打断分区下推，慎用于过滤 |
F.datediff | 日期差 | F.datediff("end", "start") | 返回天数，端点为 null 返回 null |
F.row_number | 行号 | F.row_number().over(w) | 必须配合 Window |
F.size | 数组长度 | F.size("arr") | null 返回 -1 |
F.explode | 数组炸开 | df.select(F.explode("arr")) | 一行变多行，数据量会放大 |
explode需要特别提醒：它是把数组或 map 类型的列拆成多行，数据量会成倍增长。如果你在 explode 之后又做了聚合，务必先确认 explode 后的数据规模。我曾经处理过一个字段存了 200 个标签的数组，explode 之后数据量放大 200 倍，直接把任务撑死。正确做法是先过滤标签再 explode，或者用F.size做前置校验。
6. 从逻辑到执行：算子串联时的优化思路
6.1 用 explain 看懂执行计划
写完一段 DataFrame 代码，最值得养成的习惯是打一遍df.explain(True)，看看优化器到底做了哪些处理。输出分为 Parsed、Analyzed、Optimized、Physical 四层，日常关注 Optimized 和 Physical 就够了。
Optimized 计划里能看到谓词下推、列裁剪、常量折叠是否生效。Physical 计划里能看到 join 策略、shuffle 的分区数、是否用了 Broadcast。如果 Physical 里出现了Exchange节点，说明这里有 shuffle，可以数一下有几个 Exchange，那就是这段逻辑的 shuffle 次数。
我判断一段代码好不好，一个粗略的指标就是 Exchange 节点的数量。三个以上的 Exchange 基本就有优化空间了，通常是可以通过调整算子顺序合并掉的。
6.2 分区数量的计算与调整
shuffle 后的分区数直接决定了并行度和单 task 的数据量。默认值spark.sql.shuffle.partitions=200，这个值在数据量小的时候会导致大量空 task 启动开销，在数据量大的时候又会导致单 task 数据过多。
我的估算方法是：目标单分区数据量控制在 100MB 到 200MB 之间。假设 shuffle 阶段要处理 50GB 数据，那么分区数大约取 50GB / 150MB ≈ 340，向上取整到 400 左右比较合适。同时这个数字不要超过集群总核数的 3 到 5 倍，否则 task 排队太严重。
不过现在更推荐的做法是开启 AQE 的自动合并：
spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")这样 Spark 会在运行时根据实际数据量动态合并小分区，不用你手工估算。我实测下来，开启 AQE 之后大部分任务的性能波动明显减小，尤其是数据量每天变化很大的场景。
注意：
repartition和coalesce要分清。repartition(n)会全量 shuffle，可以增可以减，数据分布均匀；coalesce(n)只能减分区，不 shuffle，是把多个分区合并到一个 task 里读，容易造成数据不均。写文件前减少小文件用coalesce，需要均匀分布用repartition。
6.3 缓存策略：什么时候该 cache
cache()本质是persist(StorageLevel.MEMORY_AND_DISK)。判断要不要缓存的标准很简单：同一个 DataFrame 是否会被多次 Action 触发。会，就缓存；不会，就别加。
我见过有人在每个中间结果上都cache()，结果内存被各种中间数据占满，真正的计算反而一直 spill 到磁盘。缓存的收益和内存占用是此消彼长的关系，只有复用次数足够多（至少两次以上）、且数据量可控时才划算。
缓存的另一个细节是别缓存太早。如果你在一个大表上先 cache 再 filter，缓存的是全量数据。正确顺序是先 filter 缩小数据量，再 cache 缩小后的结果。反过来做等于用宝贵的内存存了一堆之后用不到的数据。
用完记得unpersist()，尤其在一个长生命周期的脚本里。我有个调度任务因为忘了释放，第二天开始就频繁 GC，排查了很久才发现是缓存没清理。
7. 实际踩过的坑与速查对照
7.1 高频报错与排查路径
把这一年里遇到的高频问题整理成一张表，遇到报错先查这张表，大部分能直接定位。
| 报错/现象 | 常见原因 | 解决方向 |
|---|---|---|
AnalysisException: cannot resolve 'xxx' | 列名拼错、join 后列名冲突 | 检查 schema，join 后用 alias |
Reference 'id' is ambiguous | 两表同名列未消歧 | 用on=列表或显式 alias |
Task not serializable | 闭包里引用了不可序列化对象 | 把对象初始化放到 executor 内部 |
| 任务卡在最后一个 task | 数据倾斜 | 查 key 分布，加盐或分离热点 |
OutOfMemoryError: Java heap space | collect 太多或 collect_list 过大 | 去掉 collect，改成分批 write |
Container killed by YARN | 内存超限被外部杀掉 | 调大 executor 内存或减少单 task 数据量 |
| 结果比预期少很多 | dropna误删或 inner join 丢行 | 检查空值和 join 类型 |
| 数据重复 | join 一对多未去重 | join 前先去重或改用 semi join |
Task not serializable这个坑特别典型。原因是你传给算子的函数引用了 Driver 端的对象（比如一个数据库连接、一个配置类），Spark 需要把这个对象序列化后发到 executor，但对象本身不支持序列化。解法是把初始化逻辑放进mapPartitions里，让每个 executor 自己创建。
# 错误示范：在闭包里直接用 driver 端的对象 client = SomeClient() # 不可序列化 df.rdd.map(lambda x: client.call(x)) # 正确示范：每个分区内初始化 def process(iter): client = SomeClient() for row in iter: yield client.call(row) df.rdd.mapPartitions(process)7.2 几条不太好写进文档的经验
第一条：写 DataFrame 逻辑时，先想清楚最终输出的粒度，再倒推算子顺序。很多人是顺着数据流一条条往下写的，写到一半发现需要某个维度的汇总值，又得回头 join 一遍。如果先把目标表的粒度定下来（比如"城市 × 品类 × 天"），就知道哪些维度要保留、哪些聚合要先做，代码会简洁很多。
第二条：调试时用take(n)而不是collect()。collect()会把全量数据拉到 Driver，数据稍微大一点就是 OOM。take(10)只取前 10 行，Spark 读到够数就停，非常安全。看 schema 用printSchema()，看数据量用count()（但要意识到这会触发全量扫描）。
第三条：列裁剪永远不嫌早。select出需要的列不仅减少网络传输，列式存储还能只读需要的列文件。我有过把一个宽表 200 列在第一步就裁到 12 列的操作，整个链路的内存占用降了将近 70%。
第四条：join 键的数据类型一定要一致。一个是 int 一个是 string，Spark 会做隐式转换，但这个转换会导致无法走分区裁剪，也无法走某些优化路径，性能差好几倍。而且隐式转换在数据有脏值时会静默返回 null，join 结果莫名变少。这个坑我踩过两次，现在 join 前一定printSchema()扫一眼两边的类型。
第五条：别迷信参数调优，先看数据量。很多性能问题根本不是参数问题，而是数据量本身就超出了当前资源能处理的规模。这种情况调什么都救不了，得从业务上想办法：能不能减少数据范围？能不能改成增量计算？我见过有人为了让一个全量任务跑通，把各种参数调了一遍，折腾两天，最后发现只要把处理范围从"全历史"改成"最近三个月"，问题立刻消失。
最后分享一个我自己一直用的检查习惯：写完一段 DataFrame 代码，先拿 1% 的采样数据跑一遍完整流程，确认结果正确、算子顺序合理，再上全量。采样用df.sample(0.01)就够了，它会走一个简单的伯努利采样，代价很低。这招帮我省下了大量在集群上等待的时间，尤其是在验证聚合逻辑和 join 关系对不对的时候，小数据跑得快，错误暴露得也早。
