最近我完完整整刷了一套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, headerTrue, inferSchemaTrue)实际干活的时候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函数实战清单先查内置函数再考虑自己写UDF3.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)返回abconcat_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到8GBspark.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前面加一个EXPLAINEXPLAIN 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对比前后差异一遍下来绝对比刷十遍视频有用。