说实话SparkSQL 这两年被提起的频率高但真把它用得不出问题的人不多。我见过不少从 Hive 或者纯 Python 数据处理转到 Spark 平台的团队第一周全在跟 DataFrame 的语法较劲字符串函数拼得头晕眼花join 一跑就数据倾斜写 Hive 表之前忘了考虑文件格式最后看监控里一堆小文件只能叹口气。这篇文章就基于我自己日常在 SparkSQL 上做的最多的那些操作来写涵盖数据加载、转换、聚合、编写 SQL、写入、调优和问题排查。文中代码基于 Spark 3.x如果你还停在 2.1不建议继续用老 API尽早升上来很多省心的地方在 3.x 里才发挥得出来。1. 先搞清楚 SparkSQL 在项目里到底承担什么角色1.1 一句话定位数仓和数据湖里的主力计算引擎SparkSQL 本质上是基于 Spark 平台的一套结构化数据处理模块核心工作是让你用 SQL 或者 DataFrame API 去操作分布式数据集。它跟 Hive 很像但底层计算不再是 MapReduce而是 Spark 自己的 DAG 执行引擎。实际项目里数据从 Kafka 落到 HDFS或者从业务库同步过来第一层清洗和标准化十有八九是 Spark 做的这一层的代码往往就是 SparkSQL 的读写和转换操作。我接触到的很多网约车、电商、用户行为分析类项目场景几乎都是同一套业务数据落到 Hive 表离线加工层用 Spark 做清洗、关联、聚合再落到结果表最后供报表或大屏查询。这中间的高频动作无非是读表、过滤脏数据、字段转换、join、group by、窗口函数、写入目标表。所以“常用操作”其实并不难难的是把每一步做对、做稳。1.2 什么时候不该用 SparkSQL比会用什么更重要的是知道什么场景别用它。SparkSQL 适合大批量离线计算、批量清洗、分钟级以上的批处理。但如果是毫秒级交互查询用 Spark 是给自己找罪受延长查询延迟不说集群资源开销还大多表频繁小事务更新也不是 Spark 的强项这种场景换上 OLAP 引擎或普通关系型数据库反而合适。选错引擎的事我见过太多最典型的就是拿 Spark 跑线上接口每次查询等扫描全表接口超时后大家还以为是代码问题。1.3 与 HiveSQL、Presto、Flink SQL 的分工HiveSQL 在纯离线路线上便宜稳定但跑复杂计算时速度往往不如 SparkSQL。Presto/Trino 擅长多数据源联邦查询响应快但不是一个数据加工引擎复杂 ETL 里用它并不顺手。Flink SQL 解决的是实时流处理如果你要的是分钟级、小时级批量任务交给 SparkSQL 更合理。做架构选型时别被新概念带着跑。我就见到过某个项目把实时、离线、即席查询三套引擎都上了结果每张表要维护三份口径数据对不齐运维成本高出一截。对大多数离线数仓而言SparkSQL 加 Hive 已经够了。2. 打通 SparkSession第一次连接集群的正确姿势2.1 依赖引入与初始化操作 SparkSQL 绕不开SparkSession。在 Spark 2.0 以后它就是统一入口不需要再分别创建 SparkContext、SQLContext、HiveContext。你只需要一个对象然后所有读取、计算、注册表、SQL 都从它身上走。import org.apache.spark.sql.SparkSession val spark SparkSession .builder() .appName(spark_sql_ops_demo) .enableHiveSupport() .config(spark.sql.shuffle.partitions, 200) .config(spark.sql.adaptive.enabled, true) .getOrCreate()如果只需要本地测试不用碰集群可以加.master(local[*])。但生产环境里不要硬编码 master让提交任务时用--master参数控制代码和部署解耦方便切换测试与生产。Java 环境下的写法也类似SparkSession spark SparkSession.builder() .appName(spark_sql_ops_demo) .enableHiveSupport() .config(spark.sql.shuffle.partitions, 200) .getOrCreate();Python 用户就直接用 pysparkfrom pyspark.sql import SparkSession spark SparkSession.builder \ .appName(spark_sql_ops_demo) \ .enableHiveSupport() \ .config(spark.sql.shuffle.partitions, 200) \ .getOrCreate()很多团队成员用 Python 写业务逻辑底层跑的还是 SparkSQLAPI 口感和 Pandas 有明显差别别弄混。DataFrame 在这里是分布式对象不是 Pandas 的内存 DataFrame操作它的代价远高于本地数据所以写代码时要有分布式思维。2.2 常用配置项参考配置项不需要一次背全日常高频的也就那几个。我整理了一张表供参考配置项建议值含义与适用场景spark.sql.shuffle.partitions200 或根据集群调整shuffle 时分区数默认 200。数据量大时可以调大数据量小调小能省资源spark.sql.adaptive.enabledtrue开启 AQE动态调整执行计划spark.sql.adaptive.coalescePartitions.enabledtrue自动合并小分区spark.sql.files.maxPartitionBytes128M默认读取文件时单个分区的最大字节数spark.sql.autoBroadcastJoinThreshold10M 或调大小于阈值的小表自动广播 joinspark.sql.session.timeZoneAsia/Shanghai时区设置影响时间字段解析这些参数不要跟风乱调。默认值是基于一般场景设计的你只有先看 Spark UI 上任务的运行情况才能真正确定哪里需要调整。后面我会细说怎么通过执行计划来判断。3. DataFrame 高频操作盘点3.1 数据加载挡住脏数据的第一道门槛日常最常干的事就是把外部数据读进来。无论是 Hive 表、Parquet 文件、JSON、CSV还是 Kafka 落下来的临时文件读取时第一件事就是确认格式和 schema。用错 schema 或漏掉格式后面调试会花掉双倍时间。# 读取 Parquet 文件 df spark.read.parquet(/data/event/20240901) # 读取带表头的 CSV注意 schema 推断风险 df spark.read.option(header, true) \ .option(inferSchema, true) \ .csv(/data/raw/user_log.csv) # 读取 JSON 文件 df spark.read.json(/data/raw/user_profile.json)CSV 的inferSchema在数据量大或字段字典不规范时很容易出错。比如说金额列某些分区里全是空字符串另一分区里有数字推断结果可能变成 string后续做 sum 聚合时直接报错。稳妥做法是显式定义 schema。from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType schema StructType([ StructField(user_id, LongType(), True), StructField(city_id, StringType(), True), StructField(amount, DoubleType(), True), StructField(event_time, StringType(), True) ]) df spark.read.option(header, true) \ .schema(schema) \ .csv(/data/raw/order_20240901.csv)读取时用不用option(header, true)取决于文件内容。生产表一般建议写死 schema不要依赖 infer尤其是字段会变的数据源显式 schema 能在读取阶段挡住一部分数据质量异常。3.2 列处理与类型转换拿到原始 DataFrame 后超过一半的时间会花在字段改造上重命名、类型转换、空值处理、派生字段。from pyspark.sql import functions as F from pyspark.sql.types import DoubleType # 类型转换 df df.withColumn(amount, F.col(amount).cast(DoubleType())) # 重命名列注意多级元数据场景 df df.withColumnRenamed(orderAmount, amount) # 从时间戳字段提取日期 df df.withColumn(dt, F.to_date(F.col(event_time))) # 处理 null空值填默认值 df df.withColumn(amount, F.when(F.col(amount).isNull(), 0.0) .otherwise(F.col(amount)))有两个点要特别提醒withColumn每次调用都会返回新的 DataFrameSpark 有 lazy 执行表达式链很长也不会立刻爆炸但尽量不要在一个 DataFrame 上反复赋值尽量把一次需要新增的多列用select配合*一次搞定df df.select( *, F.to_date(F.col(event_time)).alias(dt), F.when(F.col(schema_v) 2, F.col(amount2)).otherwise(F.col(amount)).alias(real_amount) )处理金额和百分比类字段时千万不要用 float 存储做聚合很容易出现精度问题。线上金额统一用 Decimal或者转换成分单位存 long。3.3 过滤、去重与连接过滤操作看似简单但容易出问题的点在于对 null 的处理。很多人习惯写where(city_id ! 1)然而如果 city_id 本身有 null这个条件会把 null 行过滤掉因为 SQL 的三值逻辑里 null 与 1 既不相等也不不等结果 unknown。你想要保留 null 行时要写where(city_id is null or city_id ! 1)。去重是另一个高频操作。dropDuplicates和distinct的差别是前者可以指定部分列做去重基准后者是全列。日活用户这类需求我会先按 dt 分区过滤再按 user_id 去重不要在全表范围去重那是对集群资源的浪费。# 按 user_id 去重保留最新一条记录 df_unique df.orderBy(F.col(event_time).desc()) \ .dropDuplicates([user_id])连接join是最容易出性能问题的地方。两个大表做 reduce join 时如果关联键分布不均就会数据倾斜。常见做法是看关联键的基数如果某一个值占了特别多先用过滤或加盐方式打散。后面调优部分我会单独讲。# 广播小表 join避免 shuffle df_result df_big.join(F.broadcast(df_small), city_id, left)3.4 聚合与窗口函数聚合是数据分析里绕不开的步骤。简单groupBy加聚合函数人人都知道但实际场景里会有这些细节approx_count_distinct比countDistinct快几十倍适合超大基数的去重计数场景比如日活用户数误差一般可接受。groupBy后如果想加比例、排名不要拆成两步再 join用窗口函数一步到位。窗口函数是 SparkSQL 最常用、也最容易踩坑的功能。下面的例子是求每个城市订单金额前 10 的用户from pyspark.sql.window import Window window_spec Window.partitionBy(city_id).orderBy(F.col(amount).desc()) df_rank df.withColumn(rk, F.row_number().over(window_spec)) \ .filter(F.col(rk) 10)注意窗口函数的orderBy后面如果还有别的字段参与分组要写清楚 partitionBy 和 orderBy 的顺序。窗口函数里orderBy的默认排序方向是升序倒序要显式 desc。排名函数row_number、rank、dense_rank三者在并列值时的行为不同要按业务场景挑。聚合里还有个新手必坑直接在groupBy的 agg 里用自定义表达式写着写着把聚合列写进去导致非确定性结果。比如df.groupBy(city_id).agg(F.sum(F.col(amount)).alias(total_amount))这样没问题。但如果你加一个F.col(city_name)在 agg 里而它不在 groupBy 中Spark SQL 不一定立刻报错结果可能根本没意义。养成只聚合指标列分组列全部放 groupBy 的习惯。4. SQL 式开发和 SparkSQL 的优化逻辑4.1 临时视图与 createOrReplaceTempViewDataFrame API 与 SQL 可以无缝切换。你可以把 DataFrame 注册成临时视图然后写纯 SQL也可以直接用 SQL 查询一份 Hive 表再把结果映射成 DataFrame。df.createOrReplaceTempView(order_detail) result spark.sql( SELECT city_id, COUNT(*) AS order_cnt, SUM(amount) AS total_amount FROM order_detail WHERE dt 2024-09-01 GROUP BY city_id )临时视图的生命周期只在当前 SparkSession 里会话结束就消失。如果希望同一个应用内的多个任务复用可以用createGlobalTempView但生产环境我更建议直接落成中间 Hive 表或者在一个作业里把读、算、写串成一条链路靠临时表做中间结果风险更小。4.2 把复杂 SQL 拆成三步我见过不少团队喜欢写一长串 SQL一个语句里塞五六个 CTE逻辑复杂到 review 时没人知道它在干什么。日常写复杂统计时先拆再做第一步从源表里做必要的字段选择和初步过滤缩小数据量。第二步做连接和中间聚合把维度和指标对好。第三步最后一层只做结果排序、limit 或最终展示字段。这三步可以用临时视图衔接既方便看每一步的中间结果又方便 debug。比如做网约车订单分析时先按司机维度聚合订单数、总金额、高峰期占比再和司机画像表关联最后筛选指标异常的司机。拆分之后哪一步慢、哪一步数据有问题一眼就能看到。4.3 什么时候优先 DSL什么时候写 SQL我自己的习惯是简单过滤、列转换用 DataFrame DSL复杂聚合和窗口逻辑写 SQL。原因很简单DSL 更有类型提示能更早发现字段拼写错误SQL 则更贴近最终业务表达复杂逻辑写出来可读性强。唯一要克制的是不要两种风格混到一起一个作业里今天用 DS明天用 SQL后面维护的人会很痛苦。项目里最好统一约定能写 SQL 的复杂逻辑统一写 SQL简单链路用 DataFrame API。4.4 缓存与 Checkpoint缓存是调优里经常被过度使用的功能。它把中间结果存到内存或磁盘避免后续动作重复计算但缓存本身有开销和内存占用压力。什么时候缓存一条原则同一个 DataFrame 在后续被复用两次以上并且计算链条很长才考虑缓存。df_cached df.filter(...).cache() df_cached.count() # 触发实际计算真正把数据缓存起来注意光写cache()没有触发动作的话Spark 是 lazy 的不会立刻缓存。只有像count()、write这样的 action 才会真正触发。缓存的释放也要记得做要是每个任务都留一堆缓存后面任务执行时 Memory 压力会非常大。Checkpoint 相对少用主要用在长链路 DAG 太深导致血缘触发大量重算的场景。它会打断血缘把中间结果写到可靠存储。生产上如果发现某些 stage 反复重算可以考虑用 checkpoint 截断。5. 数据源读写与格式选择5.1 Hive、Parquet、ORC、JSON 怎么选SparkSQL 写数据最常遇到的选择题就是文件格式。很多同学习惯性写 JSON 或 CSV图的是可视化时看起来方便但跑数仓作业时这是灾难。格式压缩比扫描性能适合场景Parquet中高快支持列剪枝离线数仓首选ORC高快ACID 支持好Hive 大表、事务场景JSON低慢日志原始数据存储CSV低慢无内嵌 schema临时导出给业务方看新项目离线表我一般就选 Parquet。原因很直接列式存储天然适配分析型查询只需要查两列时不用扫全表再配上 snappy 压缩性能和空间都不错。如果集群 Hive 版本和 Spark 版本兼容ORC 也很好特别是 Hive 侧还要继续跑任务的情况。5.2 分区策略与写入分区字段的选择直接影响查询效率。时间字段是最常见的分区其次按业务属性比如城市、省份。选择分区字段时要避免分区度过高或过低。按天分区一般是通用做法如果数据量特别大可以按小时数据量小按月就行别为了显得“专业”硬切一大堆分区。写入 DataFrame 到 Hive 表最常用的方式是df.write.mode(overwrite) \ .partitionBy(dt) \ .format(parquet) \ .saveAsTable(dwd_order_detail)这里有个项目里常见的坑overwrite默认会覆盖整个表如果你只更新某一个分区很容易把整张表毁了。正确做法是使用动态分区覆盖只覆盖目标分区spark.conf.set(spark.sql.sources.partitionOverwriteMode, dynamic) df.write.mode(overwrite) \ .partitionBy(dt) \ .format(parquet) \ .saveAsTable(dwd_order_detail)写 Hive 表时还有一个常见操作是insertInto和saveAsTable的区别。insertInto要求目标表已经存在并且列顺序完全一致saveAsTable可以创建新表。旧项目很容易在这两个写法上栽跟头搞混之后列错位数据写进去查出来全是乱的。5.3 动态分区的设置细节动态分区能让你不用手动指定每个分区代码更简洁。但生产环境要配合适度参数调大了容易生成大批小文件。“大批小文件”是个高频痛点我之前在项目里见过一张表一天生成上千个几 KB 的小文件下游跑任务光打开文件就花了十分钟。治理思路有几个写入前用repartition控制输出文件数。用coalesce减少分区数但只能减少不能增加。定期对目标分区做文件合并用 Spark 读一遍再写一遍设置合理的目标文件大小。# 控制每个输出文件约为 256MB 左右 df.repartition(4).write.mode(overwrite).partitionBy(dt).parquet(/data/out/order_detail)分区数不要随便写先看要写入的数据量。2GB 数据用 8 个 partition比默认 200 个 partition 更合理。天天跑定时任务如果完全不关注输出文件数小文件问题早晚会爆发。5.4 与 MySQL 等外部系统的读写SparkSQL 经常要读写 MySQL比如把计算结果回写到业务库供应用层查询。df.write \ .mode(overwrite) \ .jdbc(jdbc:mysql://host:3306/db_name, agg_result, props)用 JDBC 的时候要注意如果表数据量大要趁早给相关字段建索引否则驱动全表扫描会拖慢整个任务。更重要的是一次写入的数据量和批次大小batchsize太大会导致 MySQL 端负载飙升太小的导入速度又不行。可以先压测再定。6. 性能调优要做的那些事6.1 先看执行计划再动手别一遇到慢任务就猜先执行explain()看看实际执行计划里哪些算子导致了大量数据倾斜或全表扫描。df.explain(extended) # 查看完整执行计划重点看 join 的方式是不是有 Broadcastshuffle 分区数是否合理filter 条件下推有没有生效。Spark 3.x 开启 AQE 后很多问题是运行时动态调整解决的但前提是你要理解优化的方向否则还是抓瞎。6.2 Spark 3.x 的 AQE 到底怎么帮你AQE 是 Spark 3.x 的重大改进。它能根据运行时统计信息动态改变执行计划比如把多个小分区合并、把 reduce join 自动转成 broadcast join、动态调整 join 的 shuffle 分区数。spark.conf.set(spark.sql.adaptive.enabled, true) spark.conf.set(spark.sql.adaptive.coalescePartitions.parallelismFirst, false)如果你们集群还在 Spark 2.x要把这些优化功能用好只能手工做很多操作。所以前面的建议再说一次生产环境能用 3.x 就尽量用 3.x无论性能还是 API 的一致性都好很多。6.3 三大高频优化操作第一减小数据量。过滤下沉永远是最直接的优化在源头做 where 和选列别等 join 之后再过滤。第二合理设置 shuffle 分区数。spark.sql.shuffle.partitions默认 200但是不是合适要看数据量。数据量只有几百 MB 时200 个分区会引入大量空任务反过来数据量几十 GB 时200 个分区又太少单个任务处理太多数据。第三广播小表。当维表小于 100MB 时强烈建议加broadcastjoin 时就不会触发 shuffle能省一大截时间和网络开销。具体阈值可以调spark.sql.autoBroadcastJoinThreshold但别盲目调太大广播表太大会压垮 executor 内存。6.4 数据倾斜的常用处理手段数据倾斜是跑 Spark 任务遇到的最典型问题。表现就是一个 stage 里某个 task 长时间执行其他 task 全跑完了还在等最后整个 job 超时。原因是某个 key 的数据量特别多比如某个城市订单量占 80%按城市聚合时一个 reducer 就要处理 80% 的数据。有两个常用处理手段加盐。把倾斜的 key 加上随机后缀拆成多个 key 并行聚合后再合并。这个办法写起来有点绕但效果直接。双流聚合。先把数据按原始 key 聚合一次再对倾斜 key 做随机散列二次聚合。适合聚合操作。实际开发中先定位是哪一步倾斜再看是 join 还是聚合引起的对症下药。倾斜问题在日志分析、用户行为数据里太常见了热门用户、热门商品的流量天然高度集中这块避不开但也别自己硬写先看调优方案有没有现成的函数可以用。7. 常见问题与排查技巧实录7.1 任务一直卡在某个 Stage看 Spark UI 的 stage 详情如果有大量 task 运行时间远高于平均大概率就是倾斜。这个时候先点进 task 看处理的数据量。某些 task 处理的数据量是其他 task 的几十倍那基本可以确定是数据倾斜。先去积木上的原始数据按 key 做一次 count摸清数据分布再决定加盐还是广播。还有另一种情况不是倾斜而是资源不足。看看 executor 数量、CPU 和内存的分配。集群资源被其他任务占用时Spark 会排队表现为 task 一直“等待中”这就要调整任务优先级而不是改代码。7.2 堆内存与 OOM 问题OOM 是高频故障要么是 executor 内存不足要么是驱动端内存不足。executor OOM 通常出现在聚合、join 或 cache 过多数据的时候。先看是哪个 stage 出的错再确认是数据量大还是代码引起的膨胀。常见诱因有读取时没有过滤列把几百个字段全读进来。join 前没有做预聚合导致参与 join 的数据量暴增。广播表大小超过阈值driver 和 executor 都扛不住。驱动端 OOM 常见于 collect 操作。千万不要在生产代码里对全量 DataFrame 做collect()那会把分布式数据全部拉到 driver 内存。确需抽样看结果时用limit(n).collect()或者show()。7.3 null 和脏数据带来的“薛定谔”结果很多时候任务没报错但结果不对劲细细查下来全是脏数据在捣乱。比如日期字段混进格式不同的值会导致to_date返回 null金额字段混入负数聚合后结果看起来很奇怪关联键有空白字符串join 后行数对不上。排查步骤很简单先单独对可疑字段做groupBy统计看 null 和空值占比然后看是否有多值。用describe、distinct组合着来基本能定位问题。把这些脏数据认识到再决定是过滤还是修正。7.4 常见问题速查表问题现象排查方向常见解法单个 task 执行时间明显偏长数据倾斜加盐、广播、调整 join 策略任务整体慢但无明显热点文件数过多、分区粒度过细coalesce、repartition、调整分区格式OOM 频繁内存不足或 collect 过多加内存、减少缓存、避免 collect写入结果文件数爆炸分区数设置不合理用 repartition/coalesce 控制输出数同一条 SQL 老返回不一致结果源数据有变更或动态分区冲突检查数据源、清洗逻辑、任务依赖join 结果行数暴增关联键有重复或类型不一致先做 key 去重、确认 join 类型排查问题要先看数据再看执行计划最后才改代码顺序反了容易白忙。我见过不少人一遇到问题先怀疑代码逻辑刷了半天没找到原因最后发现是上游同步脚本把时间字段格式给换了。所以有了异常第一步一定是确认输入数据的真实形态。写在后面的一点经验其实 SparkSQL 的常用操作说起来就是读、算、写三板斧难的是在真实的分布式环境里让它稳定、高效地跑完。我个人最大的体会是要把维度表和事实表的关系理清队里的规范统一、命名清晰把分区策略和文件格式在初始就定好后面会省掉大量调优和排查的时间。我今天写的这些点很多都是自己在生产环境里踩过坑后总结出来的。你也完全可以把文中的代码拿去改改用在你的日志清洗或离线报表任务里。如果后面因为数据量涨得快导致任务变慢记得回头来看调优那一节这几招绝对够用。