Spark SQL调优与UDF开发实战：从执行计划到数据倾斜

最近在补Spark SQL快速入门系列课程，正好刷到第10到15节，这几节把调优手段、常用SQL函数和UDF的开发串在一起看。以前用Spark SQL基本靠经验堆参数，数据量一上来就卡死，也不知道从哪看执行计划。学完这几节之后，我对“能跑”和“跑得快”有了更实在的认知。这次把课程要点、实际验证过的写法、以及踩过的坑整理成一篇记录，给正在入门Spark SQL，尤其对调优和UDF开发还处于“会用但说不清”阶段的朋友，一个可以直接上手的参考。

1. 课程内容整体拆解与学习路线

1.1 第10到15节到底讲了什么

这几节内容的设计逻辑其实很清晰：先解决“怎么把SQL写得更高效”，再解决“写不了或者内置函数不够用怎么办”。具体拆下来，大致有三块。

第一块是调优。课程里没有直接丢一堆参数，而是先教怎么看执行计划，再根据执行计划去调整分区数、广播阈值、开启自适应查询执行（AQE）。这个思路很关键，因为很多人一提调优就是百度参数列表，但换了环境、数据量变了，那些参数可能并不适用。看懂执行计划才是可持续发展的调优能力。

第二块是常用SQL。这部分很像一本速查手册，讲了窗口函数、行转列、日期处理和字符串处理等常用写法。每讲一个函数或关键字，课程都会配一个业务场景，比如“求每门课的排名”“取用户最近一次登录时间”，这种场景化教学比记函数签名有价值得多。

第三块是UDF。这里覆盖了临时函数、永久函数、Python UDF、Scala UDF，还简单带了一下UDAF。课程里专门强调了“不要为了写UDF而写UDF”，能用内置函数组合完成的操作尽量别写UDF，因为Spark SQL内置的优化规则对UDF基本是屏蔽状态。

1.2 我建议怎么学这几节

我的经验是：不要只看视频不动手。每看完一个小节，就把示例代码抄到自己的环境里跑一遍，再换一种相似的数据验证。比如课程讲ROW_NUMBER，我就拿了一份脱敏后的用户订单数据，把每个用户的订单按时间排序，取前三条。真正跑一遍之后，你才会发现分区子句要不要带、排序字段会不会为NULL、结果是不是稳定，这些试错都是看视频得不到的。

另外建议在本地准备一份不太大的测试数据，比如几百MB到1GB，既能快速出结果，又能反映真实集群上的行为模式。太大的测试数据会让迭代变慢，太小的数据又看不出shuffle和倾斜问题。我自己的习惯是先把某张核心表的最近一天数据导出到本地，所有验证都在这个子集上做，等逻辑通了再放大到全量。

1.3 学习后的能力提升

学完这六节，我最直观的感受是：拿到一条慢SQL，不再只靠猜。我会先EXPLAIN看有没有不必要的Exchange或者排序；遇到倾斜先定位分组键，再决定加盐还是广播。写UDF时也会先问自己“内置函数能不能做到”，这其实比会写UDF本身更重要。如果你正准备学这块内容，可以把它当成一个“从执行到调优再到自定义扩展”的闭环，下面我按三个主题展开。

2. Spark SQL调优：学会看执行计划再动手

2.1 第一步：EXPLAIN看得懂才算入门

课程在调优模块里先花了不少时间讲执行计划。当时我还觉得“这不就是打印一段文本吗”，后来才发现，能读懂这段文本，比记住几十个参数都管用。在Spark SQL里，最简单的查看方式是这样：

EXPLAIN SELECT user_id, COUNT(*) FROM orders WHERE dt = '2025-01-01' GROUP BY user_id;

新版Spark默认会打印解析后的逻辑计划、优化后的逻辑计划和物理计划。物理计划里藏着我们最关心的信息：哪里发生了Exchange、哪里需要排序、哪里用了BroadcastHashJoin还是SortMergeJoin。

提示：如果只想看最终物理计划，可以写成 EXPLAIN EXTENDED；想看某一阶段是否产生shuffle，重点找Exchange关键字。

拿到计划以后，我一般先数一下Exchange的个数。每一个Exchange都代表一次shuffle，而shuffle是Spark SQL慢的首要原因。如果本来一个小表join大表，却看到两个Exchange和一个BroadcastExchange，说明小表没有广播，大概率是广播阈值设置不合理。这时候再回去看表的大小统计，会比盲目调参数靠谱得多。

2.2 三个值得优先调整的参数

课程整理了一套“先捡软柿子捏”的调优清单，我实践下来很有效。

第一个是spark.sql.shuffle.partitions，默认是200。这个参数决定了shuffle阶段的分区数。很多人的误区是分区数越多越好，其实不是。如果数据量很小，比如只有几万条记录，200个分区意味着每个分区里只有几百条，调度开销远大于计算收益；如果数据量很大，200个分区又容易造成单个分区过大，出现OOM或倾斜。我通常的做法是先看输入数据规模和EXPLAIN后的预估大小，再按128、256或512这样逐步试。

第二个是spark.sql.autoBroadcastJoinThreshold，默认是10MB。当一张表小于这个阈值时，Spark会自动把它广播到每个Executor，避免shuffle join。实际情况中，如果业务维表是几十MB到一两百MB，可以适当调大，比如：

spark.sql.autoBroadcastJoinThreshold=209715200

但如果维表本身很大，强行广播会导致Driver端和Executors内存爆掉，得不偿失。所以这个值不能无脑调，需要先看表的实际大小，再用Spark UI观察广播前后的内存变化。

第三个是AQE，也就是自适应查询执行。在Spark 3.0以上版本可以开启：

spark.sql.adaptive.enabled=true spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.skewJoin.enabled=true

开启以后，Spark会根据运行时的实际统计动态合并小分区、优化join策略，甚至自动处理部分数据倾斜。我在某数据团队跑一个日活统计任务，开启AQE以后运行时间下降了接近40%，效果非常明显。当然，AQE也不是银弹，它依赖运行时统计信息，如果采样本身有问题，优化效果也会打折扣。

2.3 数据倾斜：定位与常用解法

数据倾斜是Spark调优绕不开的坎。第10到15节的调优部分专门讲了这个场景，核心思路是“定位分组键，再分而治之”。常见表现：某个Stage里一个Task处理的数据量远高于其他Task，任务进度一直卡在99%，其他Task都完成了，只有一两个还在跑。

我的处理步骤一般是三步。先定位到具体Stage，看是哪个Stage耗时最长，点击Spark UI里的Stage页面，查看Input Size和Records，找出数据量最大的Task。然后从SQL中反推该Stage对应的聚合或join键。最后根据键的分布特点选方案。

如果倾斜是因为某个脏值或异常值，比如user_id是空字符串或者一个超级热门ID，最简单的方法是先过滤掉或单独处理：

SELECT user_id, COUNT(*) FROM orders WHERE user_id IS NOT NULL AND user_id != '' GROUP BY user_id;

如果是因为热Key本身有意义，比如一个头部商家占了70%订单，那就可以考虑加盐。将Key加上随机前缀拆成多份，聚合后再合并：

SELECT split_part(salt_key, '_', 1) AS user_id, SUM(cnt) AS cnt FROM ( SELECT concat_ws('_', user_id, CAST(rand()*10 AS INT)) AS salt_key, COUNT(*) AS cnt FROM orders GROUP BY concat_ws('_', user_id, CAST(rand()*10 AS INT)) ) t GROUP BY split_part(salt_key, '_', 1);

这里加盐的份数要谨慎，不是越多越好。加盐之后中间结果成倍放大，如果份数太大，可能把一个小数据量任务也拖慢。我自己一般先用10份试，再根据实际Task耗时调整。另外要注意，如果SQL里还有JOIN条件，单纯给GROUP BY的Key加盐是不够的，还要考虑JOIN Key的分布。

2.4 小文件问题：写入端优化

除了运行慢，调优还有一个容易被忽视的方向：输出小文件。课程里用了一个例子：按某个业务字段动态分区写入，结果一天的数据生成了上万个小文件，下游任务读取时光打开文件就花了几分钟。这个现象在实时入仓场景尤其明显。

根治小文件，通常有两类手段。一类是写入前控制分区数。比如使用repartition或者coalesce来调整最终输出文件数量：

INSERT OVERWRITE TABLE dws_user_day SELECT /*+ REPARTITION(64) */ user_id, dt, SUM(amount) AS amount FROM dwd_orders WHERE dt = '2025-01-01' GROUP BY user_id, dt;

这里的/*+ REPARTITION(64) */是Spark 3.0引入的hint，能比较方便地控制shuffle后的分区数。如果不想改SQL，也可以用spark.sql.shuffle.partitions配一个偏小的值，效果类似，但可能影响同一个作业里其他Stage的性能。

另一类是动态分区前的“预聚合”。如果原始表的数据很分散，先缩小数据规模再写入，能显著减少小文件。比如先按天聚合，再按业务字段分区写结果。

注意：coalesce不会导致shuffle，通常用于减少分区数；但它也容易造成数据倾斜，因为多个分区被合并到一个Task里。如果要均匀分布，用repartition更好。

3. 常用SQL函数与窗口函数实战

3.1 排名三兄弟：ROW_NUMBER、RANK、DENSE_RANK

这段课程内容是“即学即用”的典型。排名函数在报表开发里太常用了，但三个函数之间的关系总有人搞混。

ROW_NUMBER()：顺序编号，相同值也区分先后，比如1、2、3。RANK()：相同值同排名，但会留下空位，比如1、1、3。DENSE_RANK()：相同值同排名，不留空位，比如1、1、2。

一个容易踩的坑：ROW_NUMBER()必须配合ORDER BY，表面上看它是稳定的，但如果排序列存在重复值，多个行之间的相对顺序并不确定。假如你用它做去重，比如WHERE rn = 1，重复行选哪一条并不保证。这个时候最好在ORDER BY里加一个唯一字段，比如订单ID或事件时间戳。

常用写法示例：

SELECT user_id, order_id, amount, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY order_time DESC) AS rn FROM orders;

如果需要返回每个用户最近两笔交易，外面包一层SELECT ... WHERE rn <= 2即可。这里建议把窗口定义具名化，方便复用：

SELECT ... FROM ( SELECT ..., ROW_NUMBER() OVER w AS rn FROM orders WINDOW w AS (PARTITION BY user_id ORDER BY order_time DESC) ) t WHERE rn <= 2;

具名窗口的好处是一旦要改排序字段，只需要改一处，不用满SQL去翻。

3.2 累计求和与LAG/LEAD

窗口函数的第二个高频应用是累计汇总和相邻行访问。累计求和经典案例是“计算每个用户截至当日的累计消费金额”：

SELECT user_id, dt, amount, SUM(amount) OVER (PARTITION BY user_id ORDER BY dt) AS cumulative_amount FROM user_orders;

要注意这里默认的窗口范围是从分区开始到当前行，但Spark支持更细粒度的行范围控制。如果你在ORDER BY字段有并列值，不加ROWS的默认范围实际上是RANGE，含义有一点差别。为了结果更可控，建议显式写范围：

SUM(amount) OVER ( PARTITION BY user_id ORDER BY dt ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ) AS cumulative_amount

LAG/LEAD通常用来求环比、间隔时长、前后事件对比。比如统计每个用户相邻两笔订单的时间间隔：

SELECT user_id, order_time, LAG(order_time, 1) OVER (PARTITION BY user_id ORDER BY order_time) AS prev_order_time, unix_timestamp(order_time) - unix_timestamp( LAG(order_time, 1) OVER (PARTITION BY user_id ORDER BY order_time) ) AS gap_seconds FROM orders;

课程里反复强调一点：LAG出来的空值是符合预期的，业务上要处理第一笔订单没有上一笔的情况，一般用COALESCE或IF补默认值。我一开始经常漏掉这一步，导致一堆NULL进入下游，最后还得回头改。

3.3 聚合技巧：行转列与collect_list

报表开发中经常会需要“把多行转成一行多列”，最常用的就是CASE WHEN + SUM：

SELECT user_id, SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS success_cnt, SUM(CASE WHEN status = 'fail' THEN 1 ELSE 0 END) AS fail_cnt FROM orders GROUP BY user_id;

如果想把多个值拼到一个字段里，可以用collect_list或collect_set：

SELECT user_id, concat_ws(',', collect_list(order_id)) AS order_id_list FROM orders GROUP BY user_id;

但要注意：collect_list不保证顺序，除非你在子查询里先排好序，然后在外层聚合。而且如果聚合结果是超大数组，容易把内存打爆，需要预估输出大小。collect_set会去重，但代价是会额外消耗内存。

COUNT(DISTINCT col)在小数据量没问题，大数据量下因为需要去重计算，成本较高。课程推荐用approx_count_distinct做近似去重，误差通常可控，速度却快很多。在UV类指标里特别实用。

3.4 日期、字符串处理以及脱敏

日期处理是SQL里最容易出格式错的点。比如把字符串转日期，通常用to_date，但要给定格式：

SELECT to_date('2025/01/15', 'yyyy/MM/dd') AS d;

如果不带上格式，Spark会按默认的yyyy-MM-dd去解析，很多埋点日志里的日期格式会返回NULL。这个坑我踩过好几次，排查的时候查数据才发现整个分区全是NULL，白白浪费半天。

字符串处理里，regexp_replace是做脱敏或清洗的利器。比如手机号中间四位脱敏：

SELECT regexp_replace('13812345678', '(\\d{3})\\d{4}(\\d{4})', '$1****$2') AS masked;

Spark SQL里的正则表达式语法基于Java，所以反斜杠要写两个。刚开始写经常漏掉这一层转义，导致匹配不上。还有一点：正则分组替换时，$1和$2在部分版本里大小写敏感，建议先在小数据量上验证。

4. 自定义UDF开发全流程

4.1 写UDF前先回答一个问题：是不是非写不可？

课程里最“反直觉”的建议是：UDF要能不用就不用。原因有三点。

第一，Spark SQL内置函数经过Catalyst优化器大量优化，而UDF是黑盒，优化器不知道它的语义，无法做谓词下推、常量折叠等优化。第二，逐行调用的UDF有大开销，Python UDF尤其明显，每条记录都要经过缓冲区序列化和反序列化。第三，UDF写多了会破坏代码的可读性，毕竟不是每个人都有自定义函数的上下文。

所以，能用regexp_extract、split、CASE WHEN、窗口函数组合解决的，就不写UDF。只有遇到内置函数确实处理不了的场景，比如复杂的解析算法、需要循环递归、依赖外部库，才考虑UDF。我自己刚学的时候特别喜欢用UDF，总感觉很“高级”，后来被调优教训过几次，才老实学会先做函数替代分析。

4.2 Python UDF：注册与类型声明

Python UDF的开发门槛最低，适合快速验证。注册方式如下：

from pyspark.sql import SparkSession from pyspark.sql.functions import udf from pyspark.sql.types import StringType spark = SparkSession.builder.appName("test_udf").getOrCreate() def mask_phone(phone): if phone is None or len(phone) != 11: return phone return phone[:3] + "****" + phone[7:] mask_phone_udf = udf(mask_phone, returnType=StringType()) df = spark.createDataFrame([("13812345678",), ("13998765432",)], ["phone"]) df.withColumn("masked", mask_phone_udf("phone")).show()

在提交到集群时，如果mask_phone依赖了某个Python文件，要用--py-files把依赖文件分发到所有节点，不然Executors上会报ModuleNotFoundError。这个错我在本地完全复现不了，后来查了半天才想起来是依赖没分发。

Spark 3.0以后，Python UDF还有一个加速版本：Pandas UDF，也叫Vectorized UDF。它会让整列数据进Pandas处理，与逐行UDF相比大幅减少了序列化开销。比如：

from pyspark.sql.functions import pandas_udf from pyspark.sql.types import StringType import pandas as pd @pandas_udf(returnType=StringType()) def mask_phone_pd(phone: pd.Series) -> pd.Series: return phone.str[:3] + "****" + phone.str[7:]

这个函数批量处理一列，速度提升很明显。但要注意Pandas UDF的输入输出都必须是Pandas Series或StructType下的多列，写不好容易报类型错误。另外，Pandas UDF在列数多的时候反而不一定比逐行UDF快，建议先benchmark再决定。

4.3 Scala UDF与UDAF示例

Scala UDF在性能上优于Python UDF，因为省掉了进程间数据来回传输。以脱敏函数为例：

import org.apache.spark.sql.expressions.UserDefinedFunction import org.apache.spark.sql.functions.udf val maskPhone: UserDefinedFunction = udf((phone: String) => { if (phone != null && phone.length == 11) phone.substring(0, 3) + "****" + phone.substring(7) else phone }) spark.udf.register("mask_phone", maskPhone)

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

SELECT mask_phone(phone) FROM user_info

如果需要做聚合类自定义函数，就要用UDAF。比如实现一个分组去重计数的UDAF，代码骨架如下：

import org.apache.spark.sql.Row import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} import org.apache.spark.sql.types._ class CountDistinctUDAF extends UserDefinedAggregateFunction { override def inputSchema: StructType = StructType(StructField("id", LongType) :: Nil) override def bufferSchema: StructType = StructType( StructField("ids", ArrayType(LongType)) :: Nil ) override def dataType: DataType = LongType override def deterministic: Boolean = true override def initialize(buffer: MutableAggregationBuffer): Unit = { buffer(0) = Seq.empty[Long] } override def update(buffer: MutableAggregationBuffer, input: Row): Unit = { val v = input.getAs[Long](0) buffer(0) = buffer.getAs[Seq[Long]](0) :+ v } override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { buffer1(0) = buffer1.getAs[Seq[Long]](0) ++ buffer2.getAs[Seq[Long]](0) } override def evaluate(buffer: Row): Long = { buffer.getAs[Seq[Long]](0).distinct.size.toLong } }

需要说明，这个实现只是为了演示UDAF的四个阶段，实际大数据量下用数组去重性能并不好，不要直接搬去生产环境。更好的生产级做法是使用Aggregator配合类型安全的Dataset API，或者在SQL里用approx_count_distinct。

打包提交时，要把代码和依赖打成jar包，再用spark-submit --jars或者--driver-class-path上传。如果是Spark SQL CLI里使用永久函数，还需要把jar路径写入CREATE FUNCTION语句。

4.4 使用SQL CLI注册永久函数

开发验证阶段一般用临时函数：

CREATE TEMPORARY FUNCTION mask_phone AS 'com.example.MaskPhoneUDF';

这里需要对应的jar已经放到--jars中。临时函数只在当前会话生效。想长期复用，可以注册到元数据库：

CREATE FUNCTION mask_phone AS 'com.example.MaskPhoneUDF' USING JAR 'hdfs:///apps/spark/udf/mask_phone.jar';

之后在同一个Spark/Hive会话里就可以直接调用。需要注意的是，各集群的jar路径写法不一样，本地文件一般要用file://或提前上传到HDFS。我当时第一次用CREATE FUNCTION没传USING JAR子句，结果只在当前会话能用，重连之后函数就丢了。

5. 实战问题与排查技巧实录

5.1 UDF在集群上找不到类或函数

本地跑得好好的，提交到集群就报ClassNotFoundException或者Function not defined。这通常是两个原因。

一个是jar包没有分发到Executor。用spark-submit --jars能解决大部分问题，但YARN Cluster模式下Driver不在提交机器上，一些依赖也要通过--conf spark.yarn.jars或--jars一并分发。另一个是临时函数注册语句没有随SQL执行，比如你用JDBC提交时只发了SQL，却没在会话里注册。

做法建议：把注册语句和业务SQL放在同一个SparkSession里跑，或封装到初始化脚本；提交时统一加上--jars和--py-files。我在某个凌晨跑批任务里就吃过这个亏，本地测试全通过，一上集群就报错，后来才意识到是driver端的临时函数注册没有同步到executors。

5.2 任务一直卡在某个Stage

这是数据倾斜的典型信号。先看不完的Stage是哪个，再点进Stage详情对比不同Task的Shuffle Read和Input大小。如果出现“长尾效应”，解决方案参考2.3节。

还有一种情况是广播表过大。如果某个维表被广播后，每个Executor都要读一次整表，广播本身也可能成为瓶颈。通过Spark UI看Executor的Memory Used可以发现端倪。此时要么把spark.sql.autoBroadcastJoinThreshold调小，强制走SortMergeJoin，要么对维表做裁剪后再广播。

我曾经遇到一个任务，慢不是因为数据多，而是因为每次启动都把一个300多MB的配置表广播到800个Executor，光广播就花了两三分钟。后来改成先过滤配置表字段再join，整体时间立刻降下来了。

5.3 SQL结果偶尔不一致

有时在本地跑出100条，在集群跑出101条，差一条。这个差异多数来自数据源本身，而不是计算错误。常见因素：

• 上游表是增量写入，作业拉取时数据还没写Complete，两次读取快照不一样；
• 用了count distinct或limit，在并行模式下本身不具备强一致性；
• 窗口排序字段不是唯一键，导致重复行的返回顺序不稳定。

排查时，先用spark.sql.files.maxPartitionBytes和分区条件锁定读取范围，再对比两次EXPLAIN的区别，重点看过滤条件下推和读取的分区列表。另外，如果任务里有INSERT OVERWRITE，还要确认下游读取时间点是否覆盖了写入窗口，避免读到半成品。

5.4 参数调优速查表

我把这几节里反复出现的参数整理成了一张表，方便直接抄作业。需要说明的是，参数没有标准答案，不同数据量和集群规模都要重新验证。

有些配置项在不同分支版本里默认值不一样，比如AQE在Spark 3.x的部分版本默认开启，在Spark 2.4里还没有。所以抄参数之前先确认Spark版本，最好用spark.version看一眼。

5.5 调试小技巧：小数据重现问题

遇到UDF逻辑不对或SQL结果奇怪时，不要直接在全量数据上调试。我的习惯是从源表过滤一个小的日期范围，比如最近一天或最近一小时，用limit或sample缩小数据量，在本地模式直接跑一遍。

小数据重现还有一个好处：可以快速调整参数组合，比如改广播阈值、改分区数，观察是否触发不同的join策略或shuffle路径。用这种打点验证的方式调优，比一次一次提交全量任务要高效得多。我曾经为了定位一个窗口函数的边界值，只取了一个用户的数据反复跑，几分钟就把问题范围缩到很小，省了很多集群资源。

6. 结合实操的个人体会

学完这几节，我自己最大的变化是：不敢再随便贴一个调优参数上去。每个参数都有它的适用场景，没有一劳永逸的配置。另一个深刻体会是，与其背各种SQL函数，不如掌握窗口函数、聚合和日期处理的底层规律。很多复杂的取数需求，拆解之后就是“分组-排序-取值”这三个动作。

如果你也正在学Spark SQL快速入门，我的建议是：尽快把执行计划阅读变成习惯。每次遇到慢查询，先EXPLAIN，再动手调参。遇到写不出逻辑，先想想是不是能拆成几层子查询配合窗口函数实现；实在不行再考虑UDF。UDF的部分不用求大求全，先写一个简单的脱敏函数跑通整个注册、调用、打包流程，之后再研究UDAF或类型安全的Aggregator。

这套内容学完之后，我后面还会去补Spark内存管理和任务调度相关的知识。因为调优做到后面，SQL本身的问题解决完，剩下的瓶颈往往在集群资源分配和存储格式上。希望这篇记录能帮正在学习中的人少走一点弯路，也欢迎有实际生产经验的同学一起交流不同的调优思路。