最近我完完整整刷了一套Spark SQL快速入门的系列课，整套课程里印象最深的就是第10讲到第15讲——调优和UDF被放在了一起讲。很多人觉得调优是高阶内容，UDF是少见需求，但实际做数据开发后你会发现，这两块几乎是日常工作中绕不开的组合：SQL写多了必然遇到性能问题，业务规则复杂了必然要写自定义函数。这篇我就把整个学习过程和实操经验串起来，讲清楚Spark SQL的核心语法、常用函数、UDF写法，以及一套真正能落地的调优思路。无论你是刚接触Spark的初学者，还是已经在写离线报表、清洗ETL的开发，这篇文章都能给你直接能抄作业的参考。

1. 入门阶段先认清：Spark SQL到底解决什么问题

1.1 为什么是SQL而不是手写RDD

很多初学者会有个疑惑：Spark底层的RDD那么灵活，什么都能做，为什么还要用Spark SQL这种看似"固定"的接口？我的理解是，数据开发工作中80%以上的任务本质上是"查表"和"转换"，而不是复杂的图计算或迭代算法。你用RDD写一个分组求和，要手动处理分区、序列化、聚合逻辑；用Spark SQL只需要几行SQL，底下的Catalyst优化器会帮你决定怎么扫描数据、怎么分区、怎么聚合，甚至会自动选择是不是要广播小表。

用一句话概括就是：Spark SQL把"怎么写分布式代码"的难题，变成了"怎么描述结果"，其余的事情交给优化器和执行引擎。对于团队协作来说，SQL天然是大家都能读懂的通用语言，数据产品经理能看，数仓工程师能写，不像RDD代码还要琢磨Scala或Java。

1.2 入门阶段就关心调优会不会太早

我当时也觉得，刚入门谈调优有点虚。但后来经历了几个实际场景才明白，调优不是优化技巧，而是理解Spark运行机制的钥匙。比如你写了一个join，跑得很慢，如果你不知道shuffle是什么，不知道数据倾斜长什么样，你连问题出在哪都判断不了；如果你不知道广播变量和分区的概念，你看Spark UI也只会一头雾水。

那套课程里把调优放在第10-15讲，恰好是在学完基础SQL和函数之后、进入真实项目之前。这个位置很合理：你已经有能力写出能跑的SQL，接下来要解决的是"怎么跑得更快、更稳"。调优的核心不是背参数，而是建立一套排查思路：先看执行计划，再观察Stage耗时，定位到底是数据倾斜、连接低效还是分区不合理，最后针对性调整。SQL调优也一样，先问"为什么慢"，再问"怎么改"。

2. 10分钟跑通核心流程：SparkSession、DataFrame与临时视图的三角关系

2.1 SparkSession的创建与配置

学习Spark SQL的第一步永远是创建SparkSession。以前的老代码里你会看到SparkContext、SQLContext、HiveContext分工明确，但现在统一入口就一个SparkSession，它把配置、数据源读写、SQL执行、UDF注册全包了。本地调试的时候，最简单的方式是下面这样：

from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("spark_sql_quickstart") \ .master("local[*]") \ .config("spark.sql.shuffle.partitions", "4") \ .getOrCreate()

这里有几个细节值得注意。master("local[*]")表示本地用所有可用核心跑，适合学习阶段；但如果是提交到集群，这个参数通常会在提交命令里指定，代码里最好不要写死。spark.sql.shuffle.partitions默认是200，也就是shuffle后默认分成200个分区，本地练习时数据量小，设成4更合适，不然你会发现明明小数据集却有一堆任务在空跑。这个参数后面调优部分还会细说，你现在先记住它是shuffle分区的开关。

2.2 DataFrame：从数据源到结构化视图

DataFrame可以理解为"带Schema的分布式表"，它比RDD多了列信息，所以Spark才能在上面做SQL级别的优化。读取数据时，我习惯先不问格式，统一用spark.read入口：

df = spark.read.json("hdfs://path/to/data.json") df = spark.read.parquet("hdfs://path/to/data.parquet") df = spark.read.csv("hdfs://path/to/data.csv", header=True, inferSchema=True)

实际干活的时候，CSV通常要留意header和inferSchema，前者决定第一行是不是列名，后者决定要不要让Spark自动推断字段类型。inferSchema在数据量大时会额外扫描一遍数据，所以生产环境里我一般会手动定义Schema，既能省一遍扫描，又能避免类型推断错位。Parquet和ORC这类列式存储不需要你操心Schema，文件本身就带了。

2.3 临时视图与SQL查询的协作模式

有了DataFrame之后，就可以注册成临时视图然后写SQL了。这是Spark SQL最常用的协作模式：

df.createOrReplaceTempView("orders") result = spark.sql(""" SELECT user_id, COUNT(*) AS order_cnt, SUM(amount) AS total_amount FROM orders GROUP BY user_id HAVING order_cnt > 5 """)

为什么非要先注册成视图再查？直接对DataFrame调用groupBy不也一样吗？实际经验告诉我，注册成临时视图有三个明显好处：一是SQL的可读性高，尤其是一段复杂的多表关联逻辑；二是调试方便，你可以在不同步骤之间反复用spark.sql查临时视图，不用重建DataFrame；三是团队协作时，业务同事可以直接看SQL逻辑，不用学DataFrame的API。

这里要提醒一个高频坑：临时视图是Session级别的，同一个SparkSession里有效，新建一个SparkSession或重启应用就没了。如果你要跨Session共享，得注册成全局临时视图（createGlobalTempView），但使用时要带global_temp前缀。还有一个坑是视图重名，如果两次都用createTempView不替换，会报"already exists"，所以代码里我习惯写createOrReplaceTempView，省心。

3. 常用SQL函数实战清单：先查内置函数，再考虑自己写UDF

3.1 聚合与窗口函数：最容易出错的边界条件

Spark SQL内置函数相当丰富，我之前整理过一张高频函数清单，基本覆盖了日常报表开发会用到的场景。聚合类的count、sum、avg、min、max没什么好说的，但有几个细节新手容易忽略：count(*)和count(1)包含NULL行，count(列名)不包含NULL行；sum遇到全NULL列返回NULL而不是0，前端展示时要记得用nvl或coalesce兜底。

窗口函数是分析场景的利器，尤其适合做排名、累计和同比环比。我最常用的组合是ROW_NUMBER()配合PARTITION BY和ORDER BY，比如取每个用户最近一笔订单：

SELECT user_id, order_id, order_time, amount FROM ( SELECT user_id, order_id, order_time, amount, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY order_time DESC) AS rn FROM orders ) t WHERE t.rn = 1

这里有个容易踩的坑：ROW_NUMBER()、RANK()、DENSE_RANK()三兄弟在处理并列值时的结果不一样。ROW_NUMBER()永远不并列，哪怕两条数据排序字段完全一样，也会随机给一个编号；RANK()会跳号，比如并列第一之后下一个是第三；DENSE_RANK()不跳号，并列第一后下一个是第二。取TopN场景建议用ROW_NUMBER()，但如果你希望并列的数据保留多条，就得用RANK()或DENSE_RANK()。

LAG和LEAD也是窗口函数里的高频工具，用来拿前一行的值或后一行的值，非常适合算环比增长。要注意ORDER BY的字段必须是唯一的，如果不唯一，计算顺序不稳定，结果就会偶尔对偶尔错。

3.2 日期时间与字符串处理：生产环境里的高频场景

日期函数的坑比聚合函数多，因为各种格式实在太多了。我常用的有to_date（字符串转日期）、date_add（加天数）、datediff（两个日期相差天数）、from_unixtime（时间戳转字符串）、unix_timestamp（字符串转时间戳）。其中from_unixtime默认格式是yyyy-MM-dd HH:mm:ss，如果你存的是毫秒时间戳，要先除以1000，否则日期会溢出成1970年附近的值。

字符串函数里，concat和concat_ws很容易混淆。concat('a', 'b')返回ab，concat_ws('-', '2024', '01', '01')返回2024-01-01，后者会用分隔符连接多个字段。处理敏感字段脱敏时，我常用substring截取和concat拼接，比如手机号中间四位打码：concat(substring(phone, 1, 3), '****', substring(phone, 8, 4))。还有一个高频场景是用regexp_replace清理脏数据，比如把字符串里的换行符、多余空格统一替换掉。

3.3 一个经典SQL实战案例：指标计算

把上面这些函数串起来，就是一个典型的日常指标开发。假设要算出每个店铺最近30天的GMV、订单量、客单价，以及环比增速：

SELECT shop_id, SUM(amount) AS gmv_30d, COUNT(DISTINCT order_id) AS order_cnt_30d, ROUND(SUM(amount) / NULLIF(COUNT(DISTINCT order_id), 0), 2) AS avg_order_value, ROUND( (SUM(amount) - SUM(CASE WHEN order_date >= DATE_SUB(CURRENT_DATE, 60) AND order_date < DATE_SUB(CURRENT_DATE, 30) THEN amount ELSE 0 END)) / NULLIF(SUM(CASE WHEN order_date >= DATE_SUB(CURRENT_DATE, 60) AND order_date < DATE_SUB(CURRENT_DATE, 30) THEN amount ELSE 0 END), 0) * 100, 2 ) AS gmv_growth_rate FROM orders WHERE order_date >= DATE_SUB(CURRENT_DATE, 30) GROUP BY shop_id

这个SQL里用了NULLIF防止除零，用CASE WHEN做条件聚合，用DATE_SUB动态计算日期范围。学习SQL不能只看函数列表，关键是学会怎么组合它们解决真实问题。我每次写这种指标SQL都会先在脑子里过一遍口径：分母是什么、分子是什么、空值怎么处理、时间范围怎么界定，想清楚了再动手。

4. UDF的正确打开方式：注册、封装与性能边界

4.1 为什么需要UDF

内置函数再丰富，也总有覆盖不到的业务规则。比如判断一个订单是否属于高风险订单，需要结合金额、频次、地区等多个字段做一套公司特有的规则判断。这种逻辑没法用纯SQL表达，或者表达出来又长又难维护，这时候就该写UDF了。

Spark SQL的UDF分几种：最简单的就是注册一个普通Python函数，用SparkSession.udf.register绑定给SQL用；也可以用@udf注解直接包装DataFrame的列操作。本质上都是把自定义处理逻辑变成SQL表达式的一部分。来看一个最朴素的例子：

from pyspark.sql.types import StringType from pyspark.sql.functions import udf def risk_level(amount, order_cnt): if amount is None or order_cnt is None: return "UNKNOWN" if amount > 10000 and order_cnt > 10: return "HIGH" elif amount > 5000: return "MEDIUM" else: return "LOW" spark.udf.register("risk_level", risk_level, StringType())

注册之后，SQL里就可以直接用了：

SELECT user_id, amount, order_cnt, risk_level(amount, order_cnt) AS risk FROM user_orders

写UDF时最容易忽略的是空值处理。SQL里的NULL一旦进入函数，很多操作会返回NULL或直接报错。我在函数第一行就习惯性地判断入参是否为空，给一个明确的兜底返回值，这样无论数据怎么脏，结果都不会崩。另外，UDF的参数和返回值类型一定要声明清楚，否则Spark需要额外做一次类型推断，性能会有损耗。

4.2 Python UDF的性能损失到底在哪

很多人用UDF的时候发现任务突然变慢了，第一反应是"集群资源不够"，实际上很可能是UDF本身把性能拖垮了。原因要从Spark的执行机制说起：原生SQL操作是直接在JVM里对二进制数据进行处理的，走的是Catalyst优化器和Tungsten执行引擎；而Python UDF需要把数据从JVM序列化后传到Python进程，Python算完再序列化传回JVM。这一来一回，每条数据都有额外的序列化和进程通信开销。

如果函数本身逻辑很复杂，比如用了正则、请求外部接口、做循环计算，那性能损耗会更明显。所以我的原则是：能用内置函数就用内置函数，迫不得已才用UDF，并且在用UDF之前先想想能不能拆成多个内置函数的组合。实际上很多"看起来必须写UDF"的逻辑，用CASE WHEN加上几个字符串函数就能拼出来。

4.3 更优的选择：Pandas UDF与内置函数优先

如果确实需要Python级别的处理，也尽量别用逐行UDF，改用Pandas UDF（也就是pandas_udf）。这类UDF的特点是让Pandas批量处理数据，一次接收一组Series，而不是一行一行传，序列化开销被大幅摊薄。比如要计算一个带业务规则的复杂评分，用Pandas UDF的写法是这样：

from pyspark.sql.functions import pandas_udf from pyspark.sql.types import StringType @pandas_udf(StringType()) def risk_level_batch(amount_series, cnt_series): # 这里拿到的是Pandas Series，可以向量化处理 return amount_series.apply(lambda x: "HIGH" if x > 10000 else "LOW")

Pandas UDF也不是银弹，它对内存的要求更高，而且调试相对麻烦。我的建议是：小数据量、一次性分析，直接写普通UDF就行；大数据量、每天跑的生产任务，优先考虑Pandas UDF或干脆绕开UDF。还有一个小技巧：UDF函数如果要依赖某个配置项或外部资源，不要在函数内部每次重新加载，尽量把初始化放在函数外部做一次，或者通过广播变量传进去，否则每处理一批数据都要加载一次外部依赖，任务只会越跑越慢。

5. 调优的四个主战场：数据倾斜、连接策略、分区粒度与内存参数

5.1 数据倾斜：症状、根因与三类处置手段

调优里遇到最多的就是数据倾斜。症状非常典型：某个Stage里绝大多数任务很快跑完，但有一两个任务卡在那里，运行时间比其他任务长几十倍，甚至直接OOM。根因就是某个key的数据量特别大，导致shuffle之后某一个分区承接了远超其他分区的数据。

对待倾斜，第一步不是改代码，而是确认哪个key有问题。可以先对疑似key做分组计数，按数量倒序排，一眼就能看出有没有"大头"。解决办法有三类：

第一类是加盐打散。对引起倾斜的key（比如一个热点商品ID）附加随机后缀，把它拆成多个子key分散到不同分区，聚合时先做一层局部聚合，再去掉后缀做全局聚合。这个方法要在SQL里手动实现，稍微复杂一些，但效果直接。

第二类是两阶段聚合。先加随机数做一次分组聚合，再对结果去掉随机数做第二次聚合。适合处理GROUP BY后的倾斜，典型的SQL优化手段。

第三类是动态调整Join策略。如果是join带来的倾斜，可以考虑把大表按倾斜key拆出来单独处理，倾斜的子集用广播方式和小表做join，不倾斜的部分走正常shuffle join，最后UNION结果。

5.2 连接策略：广播变量与小表大表处理

Spark的join策略里，最省事的是广播join：当一个小表足够小的时候，Spark把它复制到每个executor的内存里，join过程完全不走shuffle，直接把大表的每条数据和小表的本地副本匹配，速度非常快。

什么时候Spark会自动广播？这取决于spark.sql.autoBroadcastJoinThreshold参数，默认是10MB。如果小表超过这个阈值，Spark默认就走shuffle join。但阈值不是不能改：

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "209715200")

把阈值调到200MB，可以让更多"中等大小"的表走广播join。但我得提醒一句：广播表是要复制到每个executor的，如果executor很多，200MB * 200个executor就是40GB的网络和内存开销，不一定划算。所以调大阈值前，先评估集群规模和数据量。实际项目里我很少盲目调大阈值，更多是直接用代码提示，比如用broadcast函数显式声明要广播的表：

from pyspark.sql.functions import broadcast result = spark.sql("SELECT ... FROM big_table b JOIN small_table s ON ...") result = big_df.join(broadcast(small_df), "key", "left")

5.3 分区粒度：从文件数到shuffle分区的平衡

另一个影响性能的关键是分区数。分区太少，单个executor处理的数据量太大，容易OOM；分区太多，任务调度和序列化开销变大，反而不划算。spark.sql.shuffle.partitions控制的是shuffle之后的分区数量，默认200。对于小数据量的任务，200个分区明显太多，很多分区里就几条数据；对于大数据量的任务，200又可能不够。

我个人的经验是：观察每个Stage里任务的输入数据量，让单个任务处理的数据控制在100MB到200MB左右。比如某次shuffle读入的数据总量是20GB，那就设200个分区左右；如果只有2GB，设20个就够。这不是精确公式，但作为初调方向非常实用。输出结果时也要注意文件分区数量，用repartition或coalesce控制落盘文件的个数，否则可能一个查询产生几千个小文件，下游读取时反而更慢。

说到coalesce和repartition的区别：coalesce只减少分区，且尽量不产生shuffle，效率高，适合把结果合并成少量文件；repartition会重新打乱数据分布，既能增也能减分区，适合需要重新分区的场景。别用混了。

5.4 动态分配与内存参数：别再把所有内存都交给执行器

内存参数是初学者最容易乱调的地方。其实Spark内存分好几块，最重要的两块是执行内存（execution memory）和存储内存（storage memory），它们共享统一内存区域。如果你把spark.executor.memory调得很大，但shuffle读数据还是慢，很可能是执行内存不够导致频繁溢出到磁盘。

一个比较稳妥的起点是：executor内存给4GB到8GB，spark.sql.autoBroadcastJoinThreshold保持默认或适当调小，spark.sql.shuffle.partitions根据数据量动态设置。生产环境里一定要开启spark.dynamicAllocation.enabled，这样Spark可以根据负载自动调整executor数量，不用你手动估资源。

还有一个参数容易被忽略：spark.sql.files.maxPartitionBytes，它控制读取文件时单个分区的最大字节数，默认128MB。如果你读取超大文件时发现分区数过多，可以适当调大这个值，减少分区数量，降低任务调度开销。

6. 用Explain和Spark UI验证调优效果：只信数据不信感觉

6.1 从物理计划定位性能拐点

调优最忌讳"改了参数觉得很舒服"这种自我感动。每一步调整，都要有客观依据。第一步就是看执行计划。Spark SQL的执行计划分逻辑计划和物理计划，调优主要看物理计划。在SQL前面加一个EXPLAIN：

EXPLAIN SELECT t.shop_id, COUNT(*) AS order_cnt FROM orders t JOIN shops s ON t.shop_id = s.shop_id GROUP BY t.shop_id

输出里你会看到各个物理算子，比如Exchange（表示发生了shuffle）、BroadcastHashJoin或SortMergeJoin（表示join方式）、HashAggregate或SortAggregate（表示聚合方式）、Scan parquet（表示读数据）。我关注的顺序是：有没有Exchange、Exchange发生在哪个操作上、join方式是否为广播、扫描方式是否按分区裁剪了。如果看到两个大表做的是SortMergeJoin且数据量都很大，就该想想能不能用广播或者调整join顺序。

6.2 Spark UI里最有用的三个指标

执行计划是静态分析，Spark UI是你跑完任务后的回放证据。打开HistoryServer或者正在运行的Application页面，我最常看三个地方：一是Job级别的时间线，看哪个Job耗时最长，这个Job对应哪个SQL；二是Stage页面里的Shuffle Read和Shuffle Write量，如果Shuffle Read特别大，说明从上一个Stage拉取了很多数据，大概率有倾斜；三是单个Task的耗时分布，如果绝大多数Task几十秒，个别Task十几分钟，数据倾斜坐实了。

调完一轮参数后，别只看任务总时间少了就完事，还要看Spark UI里各Stage的耗时是否均衡。有一次我把一个任务的shuffle分区从200调到50，总时间确实降了，但点进Stage发现中间几个Task还是偏慢，进一步排查才意识到是某个key天然就占大头，最后用了加盐方案才彻底解决。

6.3 一次调优前后对比的完整记录

举个例子，之前做某个日志分析需求时，原始SQL从读取两个大表join到输出结果，跑了46分钟。我按照排查链路走了一遍：先看物理计划，发现两个大表走了SortMergeJoin，且没有触发动态分区裁剪；再看Spark UI，发现Shuffle Read接近80GB，明显偏大；再点开Task耗时分布，最大Task花了22分钟，中位数才6分钟，倾斜确定无疑。

随后我做了三步调整：第一步给配置表加了广播提示，虽然它有30MB不到阈值，但显式广播更保险；第二步把倾斜的key单独拆出来，用加盐的方式做两阶段join；第三步把spark.sql.shuffle.partitions从默认200调成80。改完再跑，总耗时降到11分钟，最大Task从22分钟降到3分钟。整个过程中我没动任何业务逻辑，纯粹靠执行计划和UI定位，再针对性地改。这就是调优的正确工作流——先观察，再假设，验证它，而不是把网上看到的参数挨个试一遍。

我自己实际动手之后最大的体会是：Spark SQL的语法和常用函数学起来很快，但真正拉开差距的就是对执行过程的理解。UDF是拿来解决业务灵活性的工具，调优是让工具跑得更顺的手段，而把这两者串起来的，是对数据分布、shuffle机制和Spark UI这老三样的熟练掌握。如果你也在刷Spark SQL的入门课程，我建议不要只看视频不动手，拿到那份课程里的示例数据，自己写一遍UDF，把几个调优参数挨个调着试试，再用UI对比前后差异，一遍下来绝对比刷十遍视频有用。