Spark SQL快速入门：调优、常用函数与UDF实践指南

最近我在重啃Spark SQL的快速入门，前面阶段还好，但到了第10到15节那部分，画风明显不一样了。这部分内容实际上是在讲三件事：调优、常用SQL函数、UDF开发。很多学习者也是从这一块开始产生明显的感受差异，SQL写得出来，但作业跑得慢，要优化又不知道先看哪里；UDF按网上的例子抄下来了，一上线就报错或者性能差得离谱。这篇文章我会按第10到15节的常见内容顺序，把整个调优体系、最常用的SQL函数、以及UDF的完整实践重新梳理一遍，结合我自己实际踩过的坑，给你一条可以直接照着走的路。

1. Spark SQL快速入门：先把底层思路立起来

1.1 一切从SparkSession开始，别跳过基础配置

很多Quick Start的例子都是把SparkSession当作背景板，写个spark.sql就完事了。但到了调优阶段你会发现，会话里的配置项比Redis缓存还重要，后面所有的资源调整、运行时行为，几乎都挂在SparkSession.builder上。

from pyspark.sql import SparkSession spark = (SparkSession.builder .appName("spark_sql_optimization") .config("spark.sql.adaptive.enabled", "true") .config("spark.sql.files.maxPartitionBytes", "128m") .enableHiveSupport() .getOrCreate())

注意里面这两行配置：自适应执行（Adaptive Query Execution）在Spark 3.2以后默认开启，但如果你从老版本迁移过来，建议显式设置一次；maxPartitionBytes控制读取文件时的分区粒度。快速入门阶段的练习数据量小，哪怕不改这些也能跑通，可一旦换成真实业务表，启动阶段把SparkSession维护好，后面调优能省一半事情。

1.2 DataFrame、Dataset、RDD到底怎么选

初学者最容易迷惑的，就是这三者之间的关系和选择。简单点说：

• RDD是最原始的数据抽象，只关心元素本身，没有Schema，也不做内置优化。现在除了写自定义底层逻辑，日常开发基本不需要直接操作RDD。
• DataFrame是“带Schema的RDD + 优化器”，把所有计算路径交给Catalyst优化器和Tungsten执行引擎，拖慢、误格式化、过滤下推这些事都自动处理了。
• Dataset在Java/Scala里还提供强类型编码器，能让编译期帮你查列名错误；但PySpark里没有真正对应的Dataset，Python侧基本统一走DataFrame。

所以对Spark SQL快速入门来说，重点就是：把SQL当成第一语言，DataFrame当成程序里表达查询的载体，用它来接Hive表、做临时视图、写业务转换。不要上来就纠结用RDD重写一遍，那不是调优，是给自己埋坑。

1.3 SQL与DataFrame API来回切换的“不传之秘”

同一个查询，写在spark.sql()里和用DataFrame API写，最终都会翻译成同一棵逻辑计划，性能没有本质区别。差别在于写代码的体验和维护成本。

我推荐的做法是：

1. 临时业务先写SQL，跑通逻辑再往DataFrame API迁移；
2. 需要在代码里动态拼条件、做循环的地方，用DataFrame API更安全；
3. 需要快速验证函数效果时，直接用selectExpr，这是少数既保留SQL语法又能在DataFrame里的好工具。

df = spark.table("ods.user_order") df.createOrReplaceTempView("user_order_v") # 方式一：纯SQL result = spark.sql(""" SELECT user_id, sum(amount) AS total_amount, count(DISTINCT order_id) AS order_cnt FROM user_order_v GROUP BY user_id """) # 方式二：API from pyspark.sql import functions as F result_api = df.groupBy("user_id").agg( F.sum("amount").alias("total_amount"), F.countDistinct("order_id").alias("order_cnt") )

这两种写法不仅结果一样，执行计划也基本一样。真正需要留心的不是选哪个，而是不要在切换过程中把列名搞混，尤其是SQL里用了别名以后。

2. 第10到15节的核心：Spark SQL调优究竟在调什么

2.1 第一步不是加资源，而是判断瓶颈在哪

第10到15节里的调优部分，最忌讳的就是一上来把spark.executor.memory从2G调到8G，然后满怀期待地重跑作业。实际多数慢作业的问题不在内存总量，而在于数据扫描、Shuffle量、倾斜和阶段划分不合理。

我在排查的时候一般按这个顺序看：

1. 管进多少：看Spark UI里每个Stage的Input大小，如果某次扫描就读了几百G，先想过滤能不能下推、能不能分区裁剪；
2. 管转多少：看Shuffle Write和Shuffle Read，如果Shuffle Write特别大，说明中间结果被放大，通常关联Key的分布或者数据膨胀有问题；
3. 管算多久：看特定Task执行时间有没有明显偏长，如果有几个Task红色高亮，基本就是数据倾斜。

可以用一行df.explain()把物理计划打出来，再和UI对照。很多时候，问题根本不在“参数”，而在你有没有在filter之前先做了一个全表join。这属于逻辑计划层面的错误，再加内存也没用。

2.2 Shuffle分区数，不是越大越好

踩过最多的坑就是“分区数越大并行度越高”的误解。spark.sql.shuffle.partitions默认是200，如果集群只有几十个核，200个小任务纯粹是增加调度和序列化开销；反过来，分区数太少又可能让单任务数据量过大，触发内存溢出的风险。

我通常按Executor核心数来估算：

shuffle分区数 ≈ executor数量 × 每executor核心数 × 2~3

比如20个Executor，每个4个核，我先设成160到240之间。这个值不需要每次调整，但面对数据量从几百M突然涨到几十T的情况，还是要回来试一轮。自适应执行打开以后，Spark会在运行中自动合并小分区，所以多设一点问题不大，真正怕的是数据倾斜后分区内部数据量失衡。

2.3 广播关联，这是第二个必调的优化

两表关联时，如果一张表很小（比如维度表几万行），默认Spark会在统计值满足阈值时自动广播。但当你手动执行了某些隐藏的转换，或统计信息不准时，它可能没广播，硬生生走了SortMergeJoin。

强制广播的方式很直接：

SELECT /*+ BROADCAST(dim) */ a.user_id, d.gender, a.amount FROM user_order a JOIN dim_user d ON a.user_id = d.user_id

注意广播不是越大越好。维度表超过几十MB，广播到所有Executor反而会占用大量内存并拖慢Driver的序列化时间，严重时直接OOM。可以通过spark.sql.autoBroadcastJoinThreshold设置自动广播上限，10MB是相对安全的默认值，如果维度表稳定在5MB以内，让它自动广播就行。

2.4 缓存与Checkpoint，别无条件用

df.cache()确实能避免重复计算，但它只在同一个SparkContext里对后续Action有效，并不斩断血缘关系。最快的情况是：你用同一个DataFrame接ROUND_LOG算了一遍指标，又用同一个DataFrame接ADDRESS算了一遍指标，这时候缓存它能省一次读取。

但缓存也不是免费的。缓存的表在任务结束后不会自动清理，频繁cache会占满内存，反而把后面的Shuffle挤出存储空间。我在相对重要的中间层会加df.unpersist()，保证用完就释放。

checkpoint()则更进一步，会把中间结果写到磁盘并切断血缘。这个操作代价更高，一般只有当DAG特别长、需要防止某个父Stage反复失效重算时才会用。简单任务里加了checkpoint，往往只是徒增IO。

2.5 数据倾斜，用加盐救场

第10到15节里的调优，一定绕不开数据倾斜。典型特征是：大部分Task几十秒跑完，极少数Task卡了十几分钟。原因通常是Join或GroupBy的Key严重集中。

最简单有效的思路是加盐。比如订单表按user_id关联，少数用户的数据量占到50%，可以给大表和小表同时加上一个较小的随机后缀：

-- 大表侧 SELECT user_id, concat(user_id, '_', floor(rand() * 10) % 10) AS salted_key, amount FROM order_tab -- 小表侧（先膨胀10倍） SELECT user_id, concat(user_id, '_', pos) AS salted_key FROM dim_user_tmp lateral view posexplode(split(space(9), ' ')) t AS pos, x

这样把原来集中的Key拆成10份并行处理，再进行一次去盐聚合。开启自适应执行后，Spark 3里的spark.sql.adaptive.skewJoin.enabled会自动检测并拆分倾斜分区，简单场景可以直接交给它处理。但注意它只能帮助关联场景，对GroupBy倾斜效果有限，手动加盐仍然是必修技能。

3. 常用SQL函数一批一批过，照着用就行

3.1 日期时间函数：处理时间字段的四件套

日常报表和数仓加工里，日期转换的诉求实在太高了，我总结出最常用的四组函数：

SELECT current_date() AS today, date_add(current_date(), -7) AS last_week_start, date_sub(current_date(), 7) AS last_week_date, datediff('2026-06-30', '2026-06-20') AS day_diff, months_between('2026-06-30', '2026-03-15') AS month_diff, from_unixtime(unix_timestamp('2026-06-30 10:00:00', 'yyyy-MM-dd HH:mm:ss'), 'yyyyMM') as month_id

四个核心点是：

• date_add和date_sub可以直接加减天数；
• datediff返回天数差值，因为方向问题经常写反，建议固定用“结束日期减开始日期”；
• unix_timestamp负责把字符串转时间戳，from_unixtime再把时间戳转成想要的格式；
• 年月分区字段通常用date_format(字段, 'yyyyMM')最方便，这个在后面过滤时可以直接用。

3.2 空值与条件分支：保证结果别带刺

一个容易忽略的细节是：Spark SQL里空字符串和NULL是两回事。很多统计结果对不上，就是因为源数据里既有NULL又有空字符串，直接where col = ''过滤会漏掉NULL，直接where col is null又会漏掉空串。

处理空值我比较常用：

SELECT user_id, COALESCE(phone, mobile, 'unknown') AS contact_phone, IF(age IS NULL, 0, age) AS age_filled, CASE WHEN status = 'A' THEN '正常' WHEN status = 'B' THEN '冻结' ELSE '未知' END AS status_text FROM user_profile

这里的COALESCE是最推荐的：它会从左到右返回第一个非NULL值，适合多列兜底。IF适合简单二分支，CASE WHEN适合多分支。特别注意：不要用NVL或IFNULL时把两边的类型搞混，比如IF(score IS NULL, 0, score)，如果score是double，写成整数0其实没问题；但如果是字符串情况，就得写成'0'。

3.3 字符串函数：清洗数据三板斧

Spark SQL自带丰富的字符串函数，快速入门阶段先掌握这几个：

SELECT user_id, trim(name) AS clean_name, upper(concat_ws('-', province, city)) AS area_code, split(ip_address, '\\.') AS ip_parts, substring(id_card, 7, 8) AS birthday, regexp_replace(mobile, '(\\d{3})\\d{4}(\\d{4})', '$1****$2') AS masked_mobile, get_json_object(extra_info, '$.level') AS level FROM user_profile

concat_ws拼接时可以自动跳过NULL，比concat对空值友好太多。split返回的是字符串数组，配合explode可以用在标签字段的行转列场景里。正则替换写在SQL里容易踩双反斜杠的坑，注意在Spark SQL里匹配\\.要写成'\\.'还是'\\.'? 实际案例里要多做一次转义。这类函数别看简单，清洗数据阶段能把它们用好，后面很多UDF都不需要写了。

3.4 聚合与窗口函数：分组和排序的黄金搭档

提到常用SQL，collect_list和collect_set必须单独讲。前者会把分组内内容收集成数组，后者会去重。常用于把明细拼成结构化的JSON或后续做行转列：

SELECT user_id, collect_list(order_id) AS order_list, count(*) AS order_count FROM order_detail GROUP BY user_id

窗口函数里最高频的三个是row_number()、rank()、lag()/lead()。比如取每个用户最近一笔订单：

SELECT user_id, order_id, order_time FROM ( SELECT user_id, order_id, order_time, row_number() OVER (PARTITION BY user_id ORDER BY order_time DESC) AS rn FROM order_detail ) t WHERE rn = 1

窗口函数和普通GROUP BY的区别在于，它不会把多行压成一行，而是在“不丢失明细行”的前提下计算排名和排序。这个特性在做去重、取TopN、计算同比环比时非常实用。

3.5 常用函数速查表

有时候写代码一时想不起函数名，我习惯维护一张速查表，配合记忆效果很好：

4. UDF实战：从“能跑”到“好用”

4.1 什么场景才值得写UDF

很多时候你不需要UDF，因为大部分业务逻辑都能用内置函数拼出来。第10到15节里专门讲UDF，不是说让你把所有SQL都改成UDF，而是告诉你“SQL表达不了的时候怎么扩展”。

我的判断标准很直接：

• 逻辑无法用内置函数表达；
• 多个SQL都要调用同一段复杂处理逻辑；
• 需要接入第三方库或外部资源；
• 处理过程重，但输入输出又不复杂。

如果只是把两列拼起来，再用顶层concat和coalesce能解决，就别画蛇添足。UDF写得好能提升复用性，写不好就是性能黑洞。

4.2 PySpark里的标准UDF写法

最经典的写法是注册一个Python函数，用udf包成Spark函数。比如现在要给每个用户等级打分，1到3级给0分，4到5级给10分，其他情况给20分：

from pyspark.sql.functions import udf from pyspark.sql.types import IntegerType def level_score(level): if level is None: return 0 if level in ["L1", "L2", "L3"]: return 10 if level in ["L4", "L5"]: return 20 return 0 level_score_udf = udf(level_score, IntegerType()) df = spark.table("ods.user_info") result = df.withColumn("score", level_score_udf(df["level"])) result.show()

如果想在SQL里直接调用，需要注册到SparkSession：

spark.udf.register("level_score_func", level_score, IntegerType()) spark.sql(""" SELECT user_id, level, level_score_func(level) AS score FROM ods.user_info """).show()

这里有一个很容易踩的坑：PySpark默认的Python UDF，每一行数据都会被Python解释器单独处理一遍，序列化开销巨大会导致性能骤降。数据量小的时候感受不明显，一旦跑上亿行，你会发现比纯SQL慢十几倍，这就是网上说“别用UDF”的原因。

4.3 把Python UDF改成向量化pandas_udf

同样是Python函数，换成pandas_udf和PandasSeries的写法，性能能从“逐行解释”升级成“按整列批量处理”。底层会调用Pandas对整批数据进行操作，并结合Arrow优化序列化：

import pandas as pd from pyspark.sql.functions import pandas_udf from pyspark.sql.types import IntegerType @pandas_udf(IntegerType()) def level_score_pd(level: pd.Series) -> pd.Series: # 整列处理，速度远快于逐行循环 scores = [] for lv in level: if lv is None: scores.append(0) elif lv in ["L1", "L2", "L3"]: scores.append(10) elif lv in ["L4", "L5"]: scores.append(20) else: scores.append(0) return pd.Series(scores) result = df.withColumn("score", level_score_pd(df["level"]))

能向量化尽量别用Map类型。特别是需要访问外部数据库、读配置文件的场景，批向量化只对行做业务逻辑，反正代码结构变化不大，性能却完全不同。注意pandas_udf在有GROUP BY时要用grouped_apply或agg配合，否则默认按整组传入的逻辑可能不是你想的那样。

4.4 用Scala写UDF并在Spark SQL里注册

如果项目是Scala或Java，UDF开发更直接，还能保留强类型体验。比如同一套逻辑在Scala里：

import org.apache.spark.sql.expressions.UserDefinedFunction import org.apache.spark.sql.functions.udf val levelScore: UserDefinedFunction = udf { level: String => level match { case "L1" | "L2" | "L3" => 10 case "L4" | "L5" => 20 case null => 0 case _ => 0 } } val df = spark.table("ods.user_info") val result = df.withColumn("score", levelScore(df("level"))) // 注册成SQL函数 spark.udf.register("level_score_sql", (level: String) => { level match { case "L1" | "L2" | "L3" => 10 case "L4" | "L5" => 20 case _ => 0 } }) spark.sql("SELECT user_id, level, level_score_sql(level) AS score FROM ods.user_info").show()

Scala UDF的本地闭包会比PySpark函数少一层Python解释器开销，但仍然无法获得Catalyst内部的优化。对于重逻辑，可以考虑把UserDefinedFunction写成Aggregator或使用typedLit，不过这一类内容已经超出快速入门了。

4.5 UDF开发常见坑，一次说清楚

第一，空值处理不能裸用。假使UDF输入列有NULL，如果函数内部直接len(level)或者.map(level => level.toUpperCase)，Spark基本会报空指针错误。所有UDF都应该先做NULL判断。

第二，不要在UDF里连接外部数据库。有人喜欢在UDF里查询一个MySQL表做补充，这是大坑。UDF会在每个Executor里被反复调用，等于每个任务都要建一次连接，轻则拖垮数据库，重则直接耗尽连接池。正确做法是先把这个表读成DataFrame，再用Join实现。

第三，关注UDF对序列化的影响。PySpark定义的UDF会把整个函数体当作闭包发给Executor，如果内部引用了外部大对象，每次Action都重新序列化，速度肉眼可见地变慢。尽量把依赖对象做成配置文件让Executor读取，或者用broeakcast变量广播进去。

第四，用spark.sql.functions里已有的内置函数替代UDF。这是所有优化里最见效的。比如日期加一天date_add可以做到，就不要写Python的timedelta；需求是JSON取值，就用get_json_object。

5. 快速排查与常见问题速查

5.1 Spark UI的第一眼该看哪

第10到15节的调优内容如果不讲Spark UI，等于白学。打开作业详情页后，我一般只看四个区域：

1. SQL Tab：能看到整棵物理计划树，以及每个节点输出多少行；
2. Stage列表：找到黄色或红色的Task，看它们的执行时长和Shuffle量；
3. Shuffle Read/Write：如果某Stage Read特别大，基本就是分区爆炸了；
4. Executors页：看内存和GC时长，GC时间占比高，说明对象创建太频繁。

配合UI，能看到代码逻辑和底层执行的关系，这是排查“为什么慢”最直接的手段。

5.2 用explain把执行计划拉出来看

光看UI还不够，可以在代码里直接打印执行计划：

df.explain("extended") # 输出里重点看这几个关键词： # - == Physical Plan == # - Filter，Project，Join，HashAggregate，Exchange # - SortMergeJoin 还是 BroadcastHashJoin

如果发现大表关联仍然走SortMergeJoin，而关联键是完全过滤条件明确的小结果集，就应该回去检查是不是统计信息没更新，或者忘了加BROADCAST提示。如果Exchange数量很多且分区间数据量悬殊，那也很容易判断出是Shuffle设置的问题。

5.3 报错信息速查表

5.4 一个从10-15角度整理的排查顺序

最后我给一个我平时实际用的“快速排查五步法”，基本能把大多数慢作业找出来：

1. 打开Spark UI先看有没有长尾Task，确认是不是倾斜；
2. 看Shuffle Read总量，如果任务数远大于Executor核数，检查分区数；
3. 看有没有Stage反复失败或重算，考虑是不是缺缓存/缺checkpoint；
4. 用explain看是不是SQL写得太暴力，比如可以先过滤再加join，就坚决把过滤放下推；
5. 最后再动资源参数，避免一上来就加内存。

这套顺序在快速入门的第10到15节内容里，算是我最想让你带走的核心理念：先看清计划，再动参数；先解决逻辑，再碰配置。

第10到15节如果认真跟下来，你会发现Spark SQL最值钱的能力反而不是写得多花哨，而是能用执行计划的视角审视自己的代码，能在合适的地方用合适的函数，能分清楚哪些地方该写UDF、哪些地方该绕过UDF。我个人在实际操作中的体会是：每完成一个调优，坚持把当时的执行计划和改动原因记在备注里，三个月后回头看，就是最真实的成长轨迹。遇到卡住的地方，先从Spark UI和explain里找答案，不要反复加资源试错，这个习惯比任何参数模板都管用。