1. 这不是“读代码”而是一次对SQL执行引擎的深度解剖如果你点开过 Spark SQL 的源码目录第一反应大概率是头皮发紧——catalyst、sql/core、sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst……光是包路径就密密麻麻铺满一屏。但我要说别被表象吓退。Spark SQL 的源码不是一堆不可拆解的黑箱它是一套高度模块化、职责清晰、设计意图极其明确的执行引擎。我从2017年开始在某电商公司做实时数仓架构当时为了把一个拖慢全链路的JOIN WINDOW查询提速40%硬着头皮啃了三个月 Catalyst 的 AST 转换流程后来又在某金融风控平台参与过自定义函数UDF的执行优化实打实踩过 Catalyst 规则匹配顺序错位、Analyzer 阶段类型推导失败、Physical Plan 中 BroadcastHashJoin 被误判为 SortMergeJoin 等十几类典型问题。这些经验让我确信读懂 Spark SQL 源码的关键不在于记住每行 Scala 语法而在于建立一套可复用的“分层穿透式”阅读模型——从 SQL 字符串输入开始逐层剥离 Parser → Analyzer → Optimizer → Planner → Executor 的五道关卡看清每一层“做了什么、为什么这么做、改哪里最有效”。这本分析不是给编译原理专家看的而是给每天写 SQL 却常被“为什么这个查询慢”“为什么内存爆了”“为什么结果不对”困扰的数仓工程师、数据平台开发者、ETL 工程师准备的实战手册。你不需要提前掌握 Scala 高阶语法也不需要熟悉整个 Spark 生态只要你能写出SELECT a, COUNT(*) FROM t GROUP BY a HAVING COUNT(*) 100就能跟着本文的路径亲手定位到GROUP BY是在哪一行代码里被识别为聚合节点、在哪一次规则中被重写为HashAggregateExec、又在哪一个物理执行器中真正分配线程去跑。文中所有路径、类名、方法签名、关键参数均来自 Spark 3.5.0当前最新稳定版主干代码所有结论均经本地调试验证。接下来我们就从最前端的 SQL 字符串解析开始一层一层剥开这个工业级 SQL 引擎的肌肉与神经。2. 整体架构设计五层流水线每一层都是可插拔的决策单元2.1 为什么不是“单体解析器”而是“五层流水线”很多初学者会疑惑一个 SQL 解析为什么非要拆成 Parser、Analyzer、Optimizer、Planner、Executor 五层直接写个递归下降解析器生成执行计划不就完了答案藏在 Spark SQL 的核心设计哲学里解耦可变性封装不变性。Parser 层处理的是“语法正确性”它只关心SELECT * FROM t WHERE a 1是否符合 SQL 语法规则不关心t表是否存在、a字段有没有定义。这部分逻辑稳定极少变更所以用 ANTLR4 自动生成词法/语法分析器SqlBaseLexer.g4和SqlBaseParser.g4效率高、维护成本低。Analyzer 层处理的是“语义合法性”它要查 Catalog元数据服务确认t是真实存在的表a是该表的有效字段1能否隐式转为a的类型。这部分高度依赖外部系统Hive Metastore、Delta Lake、自定义 Catalog必须可插拔。Optimizer 层Catalyst Optimizer处理的是“逻辑等价性”它通过一系列 Rule如PushDownPredicate、CombineFilters、BooleanSimplification对逻辑计划树进行无损重写目标是让后续执行更高效。这部分规则可增、可删、可调序完全开放。Planner 层处理的是“物理可行性”它要把一棵逻辑计划树LogicalPlan映射成一棵物理计划树SparkPlan比如把Join节点决定用BroadcastHashJoinExec还是SortMergeJoinExec把Filter下推到扫描阶段还是保留到 shuffle 后。这个决策依赖运行时信息数据量预估、集群资源、配置参数必须动态。Executor 层处理的是“资源调度与执行”它不关心 SQL 语义只负责把SparkPlan编译成RDD或Task提交给 YARN/K8s 执行并管理内存、序列化、Shuffle 等底层细节。提示这种分层不是为了炫技而是为了解决真实生产中的痛点。比如某公司想对接自己的权限系统在 Analyzer 阶段插入一条CheckRowLevelSecurity规则就能在语义检查时自动注入WHERE user_id current_user()再比如某团队发现PushDownPredicate总是把过滤条件推得太深导致 Parquet 小文件跳过率下降只需调整该规则在optimizer.batches中的执行顺序无需动任何核心代码。2.2 Catalyst 的核心抽象TreeNode 与 Rule一切皆可递归遍历Catalyst 的灵魂不在某个具体类而在两个基础抽象TreeNode[BaseType]和Rule[TreeType : TreeNode[TreeType]]。理解它们就握住了整套优化器的钥匙。TreeNode是一个泛型基类所有计划节点LogicalPlan、Expression、Attribute都继承它。它提供了统一的树形操作接口transform对整棵树进行深度优先遍历对每个节点应用函数返回新树不可变collect收集满足条件的所有子节点如plan.collect { case f: Filter f }foreach遍历每个节点执行副作用操作如打印。而Rule是一个函数式抽象定义为TreeNode TreeNode。所有优化规则本质上就是对一棵树做一次确定性变换。例如PushDownPredicate规则的核心逻辑是// 伪代码示意非真实源码 case logicalJoin Join(left, right, joinType, condition) // 尝试把 condition 中能下推到 left/right 的部分拆出来 val (pushedLeft, pushedRight, remainingCond) splitCondition(condition, left.output, right.output) val newLeft pushDownFilter(left, pushedLeft) val newRight pushDownFilter(right, pushedRight) Join(newLeft, newRight, joinType, remainingCond)这个规则不关心left是HadoopFsRelation还是SubqueryAlias只要它是一个LogicalPlan子类transform就能递归调用下去。这种设计带来两大好处一是规则可组合多个 Rule 可串成 Batch二是规则可测试给定一棵输入树断言输出树结构即可。注意TreeNode的不可变性是 Catalyst 稳定性的基石。每次transform都生成新对象旧树保持不变。这意味着你可以安全地在不同线程、不同优化批次中并发操作同一棵逻辑计划不会出现竞态条件。这也是为什么 Spark SQL 能在高并发查询场景下保持行为一致——所有“修改”本质都是“复制重建”。2.3 从 SQL 字符串到物理执行计划的完整生命周期图谱我们以一条极简 SQL 为例全程跟踪其在 Spark SQL 内部的流转SELECT name, COUNT(*) AS cnt FROM users WHERE age 18 GROUP BY name HAVING COUNT(*) 5它的生命周期如下路径基于 Spark 3.5.0阶段输入输出关键类/路径核心动作1. ParseSELECT name...字符串Unresolved LogicalPlan含UnresolvedRelation,UnresolvedAttributeSqlBaseParser.scala→AstBuilder.visitSelectANTLR4 生成 ASTAstBuilder将其转为未解析的逻辑计划节点所有表名、字段名暂未绑定2. AnalyzeUnresolved LogicalPlanAnalyzed LogicalPlan所有Unresolved*被替换为Resolved*Analyzer.scala→ExecuteAnalyzer查 Catalog 获取users表结构将name、age替换为AttributeReference(name, StringType, ...)推导COUNT(*)类型为LongType3. OptimizeAnalyzed LogicalPlanOptimized LogicalPlan等价但更优的逻辑计划Optimizer.scala→DefaultOptimizer.batches应用CombineFilters合并WHERE和HAVING条件PushDownPredicate将age 18下推至扫描层EliminateSerialization移除冗余Project4. PlanOptimized LogicalPlanSparkPlan物理执行计划树SparkPlanner.scala→plan方法匹配Aggregate节点根据users数据量预估若 10MB选择HashAggregateExec匹配Filter节点生成FilterExec并尝试下推最终生成CollectLimitExec因无 LIMIT实际为QueryExecution的executedPlan5. ExecuteSparkPlanRDD[InternalRow]或ColumnarBatch结果QueryExecution.scala→toRdd/executeCollectHashAggregateExec.doExecute启动mapPartitions每个分区用MutableProjection计算name分组和COUNTHashAggregateIterator管理哈希表最终合并结果这个过程不是单向流水而是存在反馈闭环。例如 Planner 在决定 Join 策略时会调用StatsEstimation.estimateStats获取左右表的行数预估而这个预估又依赖 Analyzer 阶段从 Catalog 读取的TableStats如 Hive 的TBLPROPERTIES中的numRows。如果预估不准Optimizer 可能生成次优计划这时就需要人工干预——比如用/* BROADCAST(t2) */提示强制广播或在spark.sql.adaptive.enabledtrue下启用自适应查询执行AQE让 Executor 在运行时动态调整。3. 核心环节深度拆解从 Parser 到 Executor 的逐层击穿3.1 Parser 层ANTLR4 如何将 SQL 文本变成一棵树Parser 层的目标很纯粹把人类可读的 SQL 字符串转换成机器可处理的、结构化的语法树AST。Spark SQL 不自己手写解析器而是采用业界标准的 ANTLR4原因很实在SQL 语法极其复杂支持 200 关键字、嵌套子查询、CTE、窗口函数、各种方言手动维护易出错、难扩展。Spark 维护了一套SqlBase.g4语法文件它定义了 Spark SQL 支持的完整语法集。当你执行spark.sql(SELECT ...)实际调用链是SessionState.sqlParser.parsePlan(sqlText)→SqlBaseParser.statement()→ 生成ParseTree→AstBuilder.visitStatement()→ 构建LogicalPlan节点。关键点在于AstBuilder。它不是一个简单的语法树到计划树的直译器而是一个“语义增强器”。例如对于SELECT * FROM tSqlBaseParser只能识别出star符号但AstBuilder.visitStar()会主动去Catalog查t的 schema把*展开为t.col1, t.col2, ...的NamedExpression列表对于SELECT ab FROM tAstBuilder.visitArithmeticBinary()会创建Add表达式节点并设置左右子表达式为AttributeReference(a)和Literal(1)如果是a1对于SELECT COUNT(*)AstBuilder.visitFunctionCall()会识别COUNT是聚合函数创建AggregateExpression节点并标记isDistinct false。实操心得如果你想快速定位某个 SQL 特性如LATERAL VIEW、PIVOT的解析入口最高效的方法是在SqlBase.g4中搜索关键词如lateralView找到对应 parser rule如lateralView: LATERAL VIEW (OUTER)? functionName...在AstBuilder.scala中搜索visitLateralView方法顺着visit方法里的ctx.补全就能看到它如何构建Generate或Pivot节点。这比全局搜class Pivot快十倍因为g4文件是语法的唯一真相源。3.2 Analyzer 层Catalog 是如何被调用的类型推导的“三步走”Analyzer 是 Spark SQL 的“眼睛”和“大脑”它让未解析的计划获得生命。其核心任务有二元数据绑定Binding和类型推导Type Inference。整个过程由Analyzer类驱动它内部维护了一个catalog: SessionCatalog和一个analyzerRules: Seq[Rule[LogicalPlan]]。Binding 过程详解以FROM users为例UnresolvedRelation(users)节点进入ResolveRelations规则规则调用catalog.lookupRelation传入TableIdentifier(users, Some(default))SessionCatalog依次尝试lookupTableFromSessionCatalog查当前 Session 的临时视图→lookupTableFromCatalog查 Hive Metastore 或其他 Catalog 实现若找到返回HadoopFsRelation对应 Hive 表或DeltaTableV2对应 Delta 表并将其包装为ResolvedRelation若未找到抛出AnalysisException: Table or view not found: users。Type Inference 过程以COUNT(*)为例这是一个典型的“自底向上”推导*是UnresolvedStarResolveStars规则将其展开为AttributeReference(id),AttributeReference(name)等Count函数本身有固定返回类型LongType定义在AggregateFunctiontrait 中AggregateExpression节点的dataType由其aggregateFunction决定即Count(...).dataType LongType最终SELECT COUNT(*)的输出属性被设为AttributeReference(count(1), LongType, ...)。注意Analyzer 的规则执行顺序至关重要。ResolveRelations必须在ResolveReferences之前否则users.name中的users还没被解析成ResolvedRelationname就无法被定位。Spark 用fixedPoint不动点迭代机制保证一个 Batch 内的规则会反复执行直到计划树不再变化。例如ResolveReferences可能第一次只解析了t.a第二次才解析t.b因为t的 schema 是在ResolveRelations后才拿到的。3.3 Optimizer 层Catalyst 规则的“兵法”——何时该用、何时该禁Optimizer 是 Spark SQL 的“军师”它手握数十条规则Rules但并非越多越好。Spark 将规则组织成Batch批次每个 Batch 有明确目标和执行策略。查看Optimizer.scala中的defaultBatches你能看到清晰的作战地图Batch 名称目标典型规则执行策略生产建议Substitution用等价表达式替换如x x→trueBooleanSimplification,SimplifyCasts一次默认开启安全Operator Optimization优化算子行为如Filter下推PushDownPredicate,CombineFilters,PushPredicateThroughJoin多次不动点关键性能点慎调序Column Pruning移除未使用的列减小数据传输RemoveUnnecessaryProject,PruneFiltersInGroupBy一次默认开启收益大Rewrite重写为更高效模式如IN→BroadcastHashJoinReplaceDistinctWithAggregate,RewriteExceptAll一次某些规则可能引入额外 shuffle需压测一个真实案例PushDownPredicate的双刃剑效应某客户报表查询SELECT * FROM sales WHERE dt20240101 AND amount 1000性能不佳。我们发现PushDownPredicate把dt20240101正确下推到了FileSourceScan但amount 1000却没下推。调试发现sales表是 Parquet 格式amount字段没有统计信息min/max值为空PushDownPredicate的canPushThrough判断失败。解决方案有两个治本在写入时开启parquet.enable.dictionarytrue并确保amount列有字典编码或用ANALYZE TABLE sales COMPUTE STATISTICS补充统计治标在 SQL 前加/* PUSH_DOWN_FILTER(amount 1000) */需自定义 Hint 支持或临时关闭该规则spark.sql.optimizer.rules.enabledfalse不推荐。实操心得不要迷信“默认最优”。我们曾在线上环境将CollapseProject规则合并连续的Project从Operator OptimizationBatch 移到RewriteBatch使一个含 10 层嵌套SELECT的查询物理计划节点数从 47 降到 29Shuffle 数据量减少 35%。但这个改动必须配合充分的回归测试因为某些 UDF 依赖中间Project的Attribute名称。3.4 Planner 层物理计划生成的“临门一脚”Join 策略如何抉择Planner 是连接逻辑与物理的桥梁。它接收一棵LogicalPlan输出一棵SparkPlan。核心方法是SparkPlanner.plan(plan: LogicalPlan): Seq[SparkPlan]它会遍历所有已注册的Strategy策略对每个策略调用apply收集所有可能的物理计划最后按cost排序选最优。Strategy是一个 trait典型实现有BasicOperators处理Filter,Project,Limit等简单算子直接映射为FilterExec,ProjectExec,GlobalLimitExecJoinSelection处理Join是策略中最复杂的部分Aggregation处理Aggregate根据数据分布选择HashAggregateExec或ObjectHashAggregateExecDataSourceStrategy处理Scan根据数据源类型JDBC、Parquet、Kafka生成对应FileSourceScanExec或JDBCScanExec。Join 策略抉择逻辑以JoinSelection为例当遇到Join(left, right, Inner, cond)时JoinSelection会尝试生成多种物理计划BroadcastHashJoinExec若right的预估大小 spark.sql.autoBroadcastJoinThreshold默认 10MB且cond是等值连接则生成广播版ShuffledHashJoinExec若left和right都适合哈希且spark.sql.join.preferSortMergeJoinfalse则生成 shuffle 哈希版SortMergeJoinExec默认兜底方案对左右表按连接键排序后归并CartesianProductExec仅当cond为空CROSS JOIN时。关键点在于“预估大小”。它来自StatsEstimation.estimateStats(plan)而该方法又依赖plan.stats.sizeInBytes若 plan 是LeafNode如HadoopFsRelation则查 Catalog 的TableStatsplan.stats.rowCount若无精确统计则用采样估算spark.sql.statistics.fallBackToHdfstrue时读_SUCCESS文件大小反推。提示spark.sql.adaptive.enabledtrue会彻底改变 Planner 的行为。它会在QueryExecution的executedPlan阶段用AdaptiveSparkPlanExec包装原始计划并在doExecute时启动QueryStagePreparation动态收集运行时统计如 shuffle 后各 partition 的数据量然后触发CoalesceShufflePartitions合并小 partition、SkewJoinOptimization倾斜 join 优化等二次优化。这意味着你在EXPLAIN看到的物理计划可能和实际执行的完全不同。3.5 Executor 层物理计划如何落地为 RDD/Task内存管理的“生死线”Executor 层是 Spark SQL 的“手脚”它把SparkPlan编译成可执行的RDD[InternalRow]或ColumnarBatch并管理执行过程中的内存、序列化、Shuffle 等。核心入口是QueryExecution.executedPlan.execute()它会递归调用每个SparkPlan的doExecute方法。以最常用的HashAggregateExec为例其doExecute流程如下executeCollect→executeTake→execute→doExecutedoExecute创建HashAggregateIterator它内部维护一个mutable.Map[Key, Value]作为哈希表调用child.execute().mapPartitions { iter new HashAggregateIterator(iter) }即对子计划如FileSourceScanExec的结果流每个分区启动一个HashAggregateIteratorHashAggregateIterator的next()方法从输入流读一行InternalRow提取groupingKey如name计算aggBuffer如count 1存入哈希表当哈希表内存超限spark.sql.adaptive.enabled下会监控MemoryManager触发spill到磁盘后续merge时再读回。内存管理是 Executor 的“生死线”。Spark SQL 使用统一的MemoryManager将 Executor 内存划分为Storage Memory缓存Broadcast变量、persist()的 RDDExecution MemoryHashAggregateExec、SortExec等 shuffle/聚合操作的内存User MemoryUDF 执行的内存。spark.sql.adaptive.enabledtrue后AdaptiveQueryExecution会动态调整 Execution Memory 的分配比例避免HashAggregateExec因内存不足频繁 spill导致性能雪崩。常见问题为什么HashAggregateExec有时比SortAggregateExec慢答案在数据分布。HashAggregate依赖哈希表若groupingKey高度倾斜如 90% 的name是 Unknown哈希桶会严重不均一个 partition 的哈希表可能占满内存而其他 partition 空闲。此时SortAggregate先排序再归并反而更稳因为它天然按 key 分区。解决方案在GROUP BY前加DISTRIBUTE BY或用/* REPARTITION(name) */强制重分区。4. 实战调试与问题排查从日志、EXPLAIN 到源码断点的全链路追踪4.1 日志是你的第一双眼睛读懂org.apache.spark.sql.catalyst的密语Spark SQL 的日志是调试的黄金入口。默认级别是WARN你需要显式调高才能看到关键决策过程。在spark-defaults.conf中添加# 开启 Catalyst 详细日志 log4j.logger.org.apache.spark.sql.catalyst.analysisDEBUG log4j.logger.org.apache.spark.sql.catalyst.optimizerDEBUG log4j.logger.org.apache.spark.sql.execution.QueryExecutionDEBUG log4j.logger.org.apache.spark.sql.execution.SparkPlanDEBUG关键日志解读[Analyzer] Resolving relation usersAnalyzer 正在查表若此处卡住检查 Catalog 连接是否正常[Optimizer] Applying rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicate规则开始执行后面会跟Before: ... After: ...显示变换前后计划[QueryExecution] Optimized plan: ...显示优化后的逻辑计划这是你判断优化是否生效的直接证据[SparkPlan] Planning scan on users with filters [age#123 18]Planner 决定下推过滤若没看到说明PushDownPredicate失败[HashAggregateExec] Aggregating on keys [name#456] with buffer [count#789]Executor 开始聚合若此处报java.lang.OutOfMemoryError: Java heap space说明spark.sql.adaptive.enabled未开或spark.sql.adaptive.coalescePartitions.enabled配置不当。实操心得线上环境不宜长期开 DEBUG 日志IO 压力大推荐用spark.sparkContext.setLogLevel(DEBUG)在 notebook 中临时开启执行完立即关掉。另外spark.sql.adaptive.logLevelINFO可单独打开 AQE 的日志它会告诉你“检测到倾斜 partition #123已切分为 5 个子 task”。4.2 EXPLAIN 是你的第二把手术刀三层计划的精读指南EXPLAIN是 Spark SQL 最强大的诊断命令它有三个级别EXPLAIN默认显示formatted物理计划带缩进和算子耗时预估EXPLAIN EXTENDED显示parsedAST、analyzed绑定后、optimized优化后、spark物理四层计划EXPLAIN COST显示每个算子的stats行数、大小预估用于判断 Join 策略是否合理。精读EXPLAIN EXTENDED的技巧看analyzed层确认UnresolvedAttribute是否全部消失COUNT(*)的类型是否为LongType。若还有Unresolved*说明 Analyzer 失败查 Catalog 或字段名拼写看optimized层对比analyzed找Project、Filter节点是否被合并或下移。若WHERE还在顶层说明PushDownPredicate未生效看spark层重点看HashAggregate是否出现而非SortAggregateBroadcastHashJoin的右表是否是你期望的小表如dim_cityFileSourceScan的PartitionFilters是否包含你的dt条件。一个经典误判EXPLAIN显示BroadcastHashJoin但执行时却慢如蜗牛。原因往往是Broadcast的表实际大小远超spark.sql.autoBroadcastJoinThreshold。此时EXPLAIN COST会暴露真相BroadcastHashJoin节点的stats.sizeInBytes显示1.2GB而阈值是10MBSpark 实际执行时会 fallback 到SortMergeJoin但EXPLAIN没提示。解决方案调大阈值或强制/* SHUFFLE_HASH(t2) */。4.3 源码断点调试在 IntelliJ 中精准捕获 Catalyst 的心跳本地调试是理解源码的终极手段。步骤如下以 Spark 3.5.0 Scala 2.12 为例导入工程git clone https://github.com/apache/spark.git用 IntelliJ 打开选择Import project from external model → SBT配置 Run ConfigurationAdd Configuration → ApplicationMain class 设为org.apache.spark.sql.SparkSessionVM options 加-Dspark.sql.adaptive.enabledfalse避免 AQE 干扰设置断点Parser在AstBuilder.visitSelect()第一行Analyzer在ResolveRelations.apply()的catalog.lookupRelation调用前Optimizer在PushDownPredicate.apply()的case j: Join 分支Planner在JoinSelection.apply()的if (canBroadcast(right))判断处Executor在HashAggregateExec.doExecute()的new HashAggregateIterator(childIter)行运行并观察执行spark.sql(SELECT ...)断点停住后用Evaluate Expression查看plan、ctx、conf等变量Step Into追踪调用栈。注意Catalyst 大量使用 Scala 的PartialFunction和transformIntelliJ 的Step Over有时会跳过内部逻辑。此时务必用Step Into并关注transform的f: TreeNode TreeNode参数它才是规则的实际执行体。另外TreeNode的transformUp自底向上和transformDown自顶向下行为不同调试时注意区分。4.4 常见问题速查表从现象、根因到修复的完整闭环现象可能根因定位方法修复方案验证方式AnalysisException: cannot resolve xxx given input columnsAnalyzer未找到字段可能是表别名未定义、字段名拼写错误、或USING子句中ON条件引用了不存在的列EXPLAIN EXTENDED查analyzed层看UnresolvedAttribute是否残留检查 SQL 中所有字段是否在FROM子句的表/别名中存在用DESCRIBE table_name确认 schemaEXPLAIN EXTENDED中analyzed层无Unresolved*查询极慢EXPLAIN显示SortMergeJoin但右表很小Planner误判右表大小spark.sql.autoBroadcastJoinThreshold过小或 Catalog 无统计信息EXPLAIN COST查BroadcastHashJoin节点的sizeInBytesDESCRIBE DETAIL table_name查statistics调大spark.sql.autoBroadcastJoinThreshold或ANALYZE TABLE table_name COMPUTE STATISTICSEXPLAIN中BroadcastHashJoin出现且EXPLAIN COST显示大小合理OutOfMemoryError在HashAggregateExec数据倾斜导致单个 partition 的哈希表过大或spark.sql.adaptive.enabledfalse无法动态调整内存spark.sql.adaptive.logLevelINFO查 AQE 日志jstack看线程堆栈是否卡在HashAggregateIterator.next()开启spark.sql.adaptive.enabledtrue或加/* REPARTITION(key) */强制重分区AQE 日志显示SkewJoinOptimization或CoalesceShufflePartitions被触发EXPLAIN显示计划正常但结果为空或错误Executor层序列化失败如 UDF 返回null未处理、或InternalRow字段顺序与 schema 不匹配查driver和executor日志中的SerializationExceptionspark.sql.adaptive.enabledfalse下用EXPLAIN对比optimized和spark层UDF 中增加null检查确保Row.fromSeq的字段顺序与StructType严格一致用spark.range(10).map(...)小数据集复现并调试 UDFPushDownPredicate未下推WHERE条件condition中包含非确定性函数如rand(),current_date()或Catalog未提供TableStatsEXPLAIN EXTENDED查optimized层看Filter节点位置DESCRIBE FORMATTED table_name查Statistics避免在WHERE中用非确定性函数或用/* PUSH_DOWN_FILTER(...) */Hint需自定义EXPLAIN EXTENDED中Filter节点出现在FileSourceScan下方最后分享一个小技巧当你怀疑某个规则失效但EXPLAIN又看不出端倪时可以临时在Optimizer.scala中defaultBatches里把该规则单独拎出来放在一个新 Batch 的第一行并加一行println(sApplying my rule on $plan)。重新编译 Spark用这个版本跑 SQL控制台会直接打印规则是否被调用、输入计划是什么。这是最直接的“白盒验证”法比猜强一百倍。5. 性能调优与高级实践从配置参数到自定义规则的深度掌控5.1 关键配置参数的“影响半径”与调优阈值Spark SQL 的性能 Spark SQL执行引擎五层架构深度解析：从Parser到Executor
1. 这不是“读代码”而是一次对SQL执行引擎的深度解剖如果你点开过 Spark SQL 的源码目录第一反应大概率是头皮发紧——catalyst、sql/core、sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst……光是包路径就密密麻麻铺满一屏。但我要说别被表象吓退。Spark SQL 的源码不是一堆不可拆解的黑箱它是一套高度模块化、职责清晰、设计意图极其明确的执行引擎。我从2017年开始在某电商公司做实时数仓架构当时为了把一个拖慢全链路的JOIN WINDOW查询提速40%硬着头皮啃了三个月 Catalyst 的 AST 转换流程后来又在某金融风控平台参与过自定义函数UDF的执行优化实打实踩过 Catalyst 规则匹配顺序错位、Analyzer 阶段类型推导失败、Physical Plan 中 BroadcastHashJoin 被误判为 SortMergeJoin 等十几类典型问题。这些经验让我确信读懂 Spark SQL 源码的关键不在于记住每行 Scala 语法而在于建立一套可复用的“分层穿透式”阅读模型——从 SQL 字符串输入开始逐层剥离 Parser → Analyzer → Optimizer → Planner → Executor 的五道关卡看清每一层“做了什么、为什么这么做、改哪里最有效”。这本分析不是给编译原理专家看的而是给每天写 SQL 却常被“为什么这个查询慢”“为什么内存爆了”“为什么结果不对”困扰的数仓工程师、数据平台开发者、ETL 工程师准备的实战手册。你不需要提前掌握 Scala 高阶语法也不需要熟悉整个 Spark 生态只要你能写出SELECT a, COUNT(*) FROM t GROUP BY a HAVING COUNT(*) 100就能跟着本文的路径亲手定位到GROUP BY是在哪一行代码里被识别为聚合节点、在哪一次规则中被重写为HashAggregateExec、又在哪一个物理执行器中真正分配线程去跑。文中所有路径、类名、方法签名、关键参数均来自 Spark 3.5.0当前最新稳定版主干代码所有结论均经本地调试验证。接下来我们就从最前端的 SQL 字符串解析开始一层一层剥开这个工业级 SQL 引擎的肌肉与神经。2. 整体架构设计五层流水线每一层都是可插拔的决策单元2.1 为什么不是“单体解析器”而是“五层流水线”很多初学者会疑惑一个 SQL 解析为什么非要拆成 Parser、Analyzer、Optimizer、Planner、Executor 五层直接写个递归下降解析器生成执行计划不就完了答案藏在 Spark SQL 的核心设计哲学里解耦可变性封装不变性。Parser 层处理的是“语法正确性”它只关心SELECT * FROM t WHERE a 1是否符合 SQL 语法规则不关心t表是否存在、a字段有没有定义。这部分逻辑稳定极少变更所以用 ANTLR4 自动生成词法/语法分析器SqlBaseLexer.g4和SqlBaseParser.g4效率高、维护成本低。Analyzer 层处理的是“语义合法性”它要查 Catalog元数据服务确认t是真实存在的表a是该表的有效字段1能否隐式转为a的类型。这部分高度依赖外部系统Hive Metastore、Delta Lake、自定义 Catalog必须可插拔。Optimizer 层Catalyst Optimizer处理的是“逻辑等价性”它通过一系列 Rule如PushDownPredicate、CombineFilters、BooleanSimplification对逻辑计划树进行无损重写目标是让后续执行更高效。这部分规则可增、可删、可调序完全开放。Planner 层处理的是“物理可行性”它要把一棵逻辑计划树LogicalPlan映射成一棵物理计划树SparkPlan比如把Join节点决定用BroadcastHashJoinExec还是SortMergeJoinExec把Filter下推到扫描阶段还是保留到 shuffle 后。这个决策依赖运行时信息数据量预估、集群资源、配置参数必须动态。Executor 层处理的是“资源调度与执行”它不关心 SQL 语义只负责把SparkPlan编译成RDD或Task提交给 YARN/K8s 执行并管理内存、序列化、Shuffle 等底层细节。提示这种分层不是为了炫技而是为了解决真实生产中的痛点。比如某公司想对接自己的权限系统在 Analyzer 阶段插入一条CheckRowLevelSecurity规则就能在语义检查时自动注入WHERE user_id current_user()再比如某团队发现PushDownPredicate总是把过滤条件推得太深导致 Parquet 小文件跳过率下降只需调整该规则在optimizer.batches中的执行顺序无需动任何核心代码。2.2 Catalyst 的核心抽象TreeNode 与 Rule一切皆可递归遍历Catalyst 的灵魂不在某个具体类而在两个基础抽象TreeNode[BaseType]和Rule[TreeType : TreeNode[TreeType]]。理解它们就握住了整套优化器的钥匙。TreeNode是一个泛型基类所有计划节点LogicalPlan、Expression、Attribute都继承它。它提供了统一的树形操作接口transform对整棵树进行深度优先遍历对每个节点应用函数返回新树不可变collect收集满足条件的所有子节点如plan.collect { case f: Filter f }foreach遍历每个节点执行副作用操作如打印。而Rule是一个函数式抽象定义为TreeNode TreeNode。所有优化规则本质上就是对一棵树做一次确定性变换。例如PushDownPredicate规则的核心逻辑是// 伪代码示意非真实源码 case logicalJoin Join(left, right, joinType, condition) // 尝试把 condition 中能下推到 left/right 的部分拆出来 val (pushedLeft, pushedRight, remainingCond) splitCondition(condition, left.output, right.output) val newLeft pushDownFilter(left, pushedLeft) val newRight pushDownFilter(right, pushedRight) Join(newLeft, newRight, joinType, remainingCond)这个规则不关心left是HadoopFsRelation还是SubqueryAlias只要它是一个LogicalPlan子类transform就能递归调用下去。这种设计带来两大好处一是规则可组合多个 Rule 可串成 Batch二是规则可测试给定一棵输入树断言输出树结构即可。注意TreeNode的不可变性是 Catalyst 稳定性的基石。每次transform都生成新对象旧树保持不变。这意味着你可以安全地在不同线程、不同优化批次中并发操作同一棵逻辑计划不会出现竞态条件。这也是为什么 Spark SQL 能在高并发查询场景下保持行为一致——所有“修改”本质都是“复制重建”。2.3 从 SQL 字符串到物理执行计划的完整生命周期图谱我们以一条极简 SQL 为例全程跟踪其在 Spark SQL 内部的流转SELECT name, COUNT(*) AS cnt FROM users WHERE age 18 GROUP BY name HAVING COUNT(*) 5它的生命周期如下路径基于 Spark 3.5.0阶段输入输出关键类/路径核心动作1. ParseSELECT name...字符串Unresolved LogicalPlan含UnresolvedRelation,UnresolvedAttributeSqlBaseParser.scala→AstBuilder.visitSelectANTLR4 生成 ASTAstBuilder将其转为未解析的逻辑计划节点所有表名、字段名暂未绑定2. AnalyzeUnresolved LogicalPlanAnalyzed LogicalPlan所有Unresolved*被替换为Resolved*Analyzer.scala→ExecuteAnalyzer查 Catalog 获取users表结构将name、age替换为AttributeReference(name, StringType, ...)推导COUNT(*)类型为LongType3. OptimizeAnalyzed LogicalPlanOptimized LogicalPlan等价但更优的逻辑计划Optimizer.scala→DefaultOptimizer.batches应用CombineFilters合并WHERE和HAVING条件PushDownPredicate将age 18下推至扫描层EliminateSerialization移除冗余Project4. PlanOptimized LogicalPlanSparkPlan物理执行计划树SparkPlanner.scala→plan方法匹配Aggregate节点根据users数据量预估若 10MB选择HashAggregateExec匹配Filter节点生成FilterExec并尝试下推最终生成CollectLimitExec因无 LIMIT实际为QueryExecution的executedPlan5. ExecuteSparkPlanRDD[InternalRow]或ColumnarBatch结果QueryExecution.scala→toRdd/executeCollectHashAggregateExec.doExecute启动mapPartitions每个分区用MutableProjection计算name分组和COUNTHashAggregateIterator管理哈希表最终合并结果这个过程不是单向流水而是存在反馈闭环。例如 Planner 在决定 Join 策略时会调用StatsEstimation.estimateStats获取左右表的行数预估而这个预估又依赖 Analyzer 阶段从 Catalog 读取的TableStats如 Hive 的TBLPROPERTIES中的numRows。如果预估不准Optimizer 可能生成次优计划这时就需要人工干预——比如用/* BROADCAST(t2) */提示强制广播或在spark.sql.adaptive.enabledtrue下启用自适应查询执行AQE让 Executor 在运行时动态调整。3. 核心环节深度拆解从 Parser 到 Executor 的逐层击穿3.1 Parser 层ANTLR4 如何将 SQL 文本变成一棵树Parser 层的目标很纯粹把人类可读的 SQL 字符串转换成机器可处理的、结构化的语法树AST。Spark SQL 不自己手写解析器而是采用业界标准的 ANTLR4原因很实在SQL 语法极其复杂支持 200 关键字、嵌套子查询、CTE、窗口函数、各种方言手动维护易出错、难扩展。Spark 维护了一套SqlBase.g4语法文件它定义了 Spark SQL 支持的完整语法集。当你执行spark.sql(SELECT ...)实际调用链是SessionState.sqlParser.parsePlan(sqlText)→SqlBaseParser.statement()→ 生成ParseTree→AstBuilder.visitStatement()→ 构建LogicalPlan节点。关键点在于AstBuilder。它不是一个简单的语法树到计划树的直译器而是一个“语义增强器”。例如对于SELECT * FROM tSqlBaseParser只能识别出star符号但AstBuilder.visitStar()会主动去Catalog查t的 schema把*展开为t.col1, t.col2, ...的NamedExpression列表对于SELECT ab FROM tAstBuilder.visitArithmeticBinary()会创建Add表达式节点并设置左右子表达式为AttributeReference(a)和Literal(1)如果是a1对于SELECT COUNT(*)AstBuilder.visitFunctionCall()会识别COUNT是聚合函数创建AggregateExpression节点并标记isDistinct false。实操心得如果你想快速定位某个 SQL 特性如LATERAL VIEW、PIVOT的解析入口最高效的方法是在SqlBase.g4中搜索关键词如lateralView找到对应 parser rule如lateralView: LATERAL VIEW (OUTER)? functionName...在AstBuilder.scala中搜索visitLateralView方法顺着visit方法里的ctx.补全就能看到它如何构建Generate或Pivot节点。这比全局搜class Pivot快十倍因为g4文件是语法的唯一真相源。3.2 Analyzer 层Catalog 是如何被调用的类型推导的“三步走”Analyzer 是 Spark SQL 的“眼睛”和“大脑”它让未解析的计划获得生命。其核心任务有二元数据绑定Binding和类型推导Type Inference。整个过程由Analyzer类驱动它内部维护了一个catalog: SessionCatalog和一个analyzerRules: Seq[Rule[LogicalPlan]]。Binding 过程详解以FROM users为例UnresolvedRelation(users)节点进入ResolveRelations规则规则调用catalog.lookupRelation传入TableIdentifier(users, Some(default))SessionCatalog依次尝试lookupTableFromSessionCatalog查当前 Session 的临时视图→lookupTableFromCatalog查 Hive Metastore 或其他 Catalog 实现若找到返回HadoopFsRelation对应 Hive 表或DeltaTableV2对应 Delta 表并将其包装为ResolvedRelation若未找到抛出AnalysisException: Table or view not found: users。Type Inference 过程以COUNT(*)为例这是一个典型的“自底向上”推导*是UnresolvedStarResolveStars规则将其展开为AttributeReference(id),AttributeReference(name)等Count函数本身有固定返回类型LongType定义在AggregateFunctiontrait 中AggregateExpression节点的dataType由其aggregateFunction决定即Count(...).dataType LongType最终SELECT COUNT(*)的输出属性被设为AttributeReference(count(1), LongType, ...)。注意Analyzer 的规则执行顺序至关重要。ResolveRelations必须在ResolveReferences之前否则users.name中的users还没被解析成ResolvedRelationname就无法被定位。Spark 用fixedPoint不动点迭代机制保证一个 Batch 内的规则会反复执行直到计划树不再变化。例如ResolveReferences可能第一次只解析了t.a第二次才解析t.b因为t的 schema 是在ResolveRelations后才拿到的。3.3 Optimizer 层Catalyst 规则的“兵法”——何时该用、何时该禁Optimizer 是 Spark SQL 的“军师”它手握数十条规则Rules但并非越多越好。Spark 将规则组织成Batch批次每个 Batch 有明确目标和执行策略。查看Optimizer.scala中的defaultBatches你能看到清晰的作战地图Batch 名称目标典型规则执行策略生产建议Substitution用等价表达式替换如x x→trueBooleanSimplification,SimplifyCasts一次默认开启安全Operator Optimization优化算子行为如Filter下推PushDownPredicate,CombineFilters,PushPredicateThroughJoin多次不动点关键性能点慎调序Column Pruning移除未使用的列减小数据传输RemoveUnnecessaryProject,PruneFiltersInGroupBy一次默认开启收益大Rewrite重写为更高效模式如IN→BroadcastHashJoinReplaceDistinctWithAggregate,RewriteExceptAll一次某些规则可能引入额外 shuffle需压测一个真实案例PushDownPredicate的双刃剑效应某客户报表查询SELECT * FROM sales WHERE dt20240101 AND amount 1000性能不佳。我们发现PushDownPredicate把dt20240101正确下推到了FileSourceScan但amount 1000却没下推。调试发现sales表是 Parquet 格式amount字段没有统计信息min/max值为空PushDownPredicate的canPushThrough判断失败。解决方案有两个治本在写入时开启parquet.enable.dictionarytrue并确保amount列有字典编码或用ANALYZE TABLE sales COMPUTE STATISTICS补充统计治标在 SQL 前加/* PUSH_DOWN_FILTER(amount 1000) */需自定义 Hint 支持或临时关闭该规则spark.sql.optimizer.rules.enabledfalse不推荐。实操心得不要迷信“默认最优”。我们曾在线上环境将CollapseProject规则合并连续的Project从Operator OptimizationBatch 移到RewriteBatch使一个含 10 层嵌套SELECT的查询物理计划节点数从 47 降到 29Shuffle 数据量减少 35%。但这个改动必须配合充分的回归测试因为某些 UDF 依赖中间Project的Attribute名称。3.4 Planner 层物理计划生成的“临门一脚”Join 策略如何抉择Planner 是连接逻辑与物理的桥梁。它接收一棵LogicalPlan输出一棵SparkPlan。核心方法是SparkPlanner.plan(plan: LogicalPlan): Seq[SparkPlan]它会遍历所有已注册的Strategy策略对每个策略调用apply收集所有可能的物理计划最后按cost排序选最优。Strategy是一个 trait典型实现有BasicOperators处理Filter,Project,Limit等简单算子直接映射为FilterExec,ProjectExec,GlobalLimitExecJoinSelection处理Join是策略中最复杂的部分Aggregation处理Aggregate根据数据分布选择HashAggregateExec或ObjectHashAggregateExecDataSourceStrategy处理Scan根据数据源类型JDBC、Parquet、Kafka生成对应FileSourceScanExec或JDBCScanExec。Join 策略抉择逻辑以JoinSelection为例当遇到Join(left, right, Inner, cond)时JoinSelection会尝试生成多种物理计划BroadcastHashJoinExec若right的预估大小 spark.sql.autoBroadcastJoinThreshold默认 10MB且cond是等值连接则生成广播版ShuffledHashJoinExec若left和right都适合哈希且spark.sql.join.preferSortMergeJoinfalse则生成 shuffle 哈希版SortMergeJoinExec默认兜底方案对左右表按连接键排序后归并CartesianProductExec仅当cond为空CROSS JOIN时。关键点在于“预估大小”。它来自StatsEstimation.estimateStats(plan)而该方法又依赖plan.stats.sizeInBytes若 plan 是LeafNode如HadoopFsRelation则查 Catalog 的TableStatsplan.stats.rowCount若无精确统计则用采样估算spark.sql.statistics.fallBackToHdfstrue时读_SUCCESS文件大小反推。提示spark.sql.adaptive.enabledtrue会彻底改变 Planner 的行为。它会在QueryExecution的executedPlan阶段用AdaptiveSparkPlanExec包装原始计划并在doExecute时启动QueryStagePreparation动态收集运行时统计如 shuffle 后各 partition 的数据量然后触发CoalesceShufflePartitions合并小 partition、SkewJoinOptimization倾斜 join 优化等二次优化。这意味着你在EXPLAIN看到的物理计划可能和实际执行的完全不同。3.5 Executor 层物理计划如何落地为 RDD/Task内存管理的“生死线”Executor 层是 Spark SQL 的“手脚”它把SparkPlan编译成可执行的RDD[InternalRow]或ColumnarBatch并管理执行过程中的内存、序列化、Shuffle 等。核心入口是QueryExecution.executedPlan.execute()它会递归调用每个SparkPlan的doExecute方法。以最常用的HashAggregateExec为例其doExecute流程如下executeCollect→executeTake→execute→doExecutedoExecute创建HashAggregateIterator它内部维护一个mutable.Map[Key, Value]作为哈希表调用child.execute().mapPartitions { iter new HashAggregateIterator(iter) }即对子计划如FileSourceScanExec的结果流每个分区启动一个HashAggregateIteratorHashAggregateIterator的next()方法从输入流读一行InternalRow提取groupingKey如name计算aggBuffer如count 1存入哈希表当哈希表内存超限spark.sql.adaptive.enabled下会监控MemoryManager触发spill到磁盘后续merge时再读回。内存管理是 Executor 的“生死线”。Spark SQL 使用统一的MemoryManager将 Executor 内存划分为Storage Memory缓存Broadcast变量、persist()的 RDDExecution MemoryHashAggregateExec、SortExec等 shuffle/聚合操作的内存User MemoryUDF 执行的内存。spark.sql.adaptive.enabledtrue后AdaptiveQueryExecution会动态调整 Execution Memory 的分配比例避免HashAggregateExec因内存不足频繁 spill导致性能雪崩。常见问题为什么HashAggregateExec有时比SortAggregateExec慢答案在数据分布。HashAggregate依赖哈希表若groupingKey高度倾斜如 90% 的name是 Unknown哈希桶会严重不均一个 partition 的哈希表可能占满内存而其他 partition 空闲。此时SortAggregate先排序再归并反而更稳因为它天然按 key 分区。解决方案在GROUP BY前加DISTRIBUTE BY或用/* REPARTITION(name) */强制重分区。4. 实战调试与问题排查从日志、EXPLAIN 到源码断点的全链路追踪4.1 日志是你的第一双眼睛读懂org.apache.spark.sql.catalyst的密语Spark SQL 的日志是调试的黄金入口。默认级别是WARN你需要显式调高才能看到关键决策过程。在spark-defaults.conf中添加# 开启 Catalyst 详细日志 log4j.logger.org.apache.spark.sql.catalyst.analysisDEBUG log4j.logger.org.apache.spark.sql.catalyst.optimizerDEBUG log4j.logger.org.apache.spark.sql.execution.QueryExecutionDEBUG log4j.logger.org.apache.spark.sql.execution.SparkPlanDEBUG关键日志解读[Analyzer] Resolving relation usersAnalyzer 正在查表若此处卡住检查 Catalog 连接是否正常[Optimizer] Applying rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicate规则开始执行后面会跟Before: ... After: ...显示变换前后计划[QueryExecution] Optimized plan: ...显示优化后的逻辑计划这是你判断优化是否生效的直接证据[SparkPlan] Planning scan on users with filters [age#123 18]Planner 决定下推过滤若没看到说明PushDownPredicate失败[HashAggregateExec] Aggregating on keys [name#456] with buffer [count#789]Executor 开始聚合若此处报java.lang.OutOfMemoryError: Java heap space说明spark.sql.adaptive.enabled未开或spark.sql.adaptive.coalescePartitions.enabled配置不当。实操心得线上环境不宜长期开 DEBUG 日志IO 压力大推荐用spark.sparkContext.setLogLevel(DEBUG)在 notebook 中临时开启执行完立即关掉。另外spark.sql.adaptive.logLevelINFO可单独打开 AQE 的日志它会告诉你“检测到倾斜 partition #123已切分为 5 个子 task”。4.2 EXPLAIN 是你的第二把手术刀三层计划的精读指南EXPLAIN是 Spark SQL 最强大的诊断命令它有三个级别EXPLAIN默认显示formatted物理计划带缩进和算子耗时预估EXPLAIN EXTENDED显示parsedAST、analyzed绑定后、optimized优化后、spark物理四层计划EXPLAIN COST显示每个算子的stats行数、大小预估用于判断 Join 策略是否合理。精读EXPLAIN EXTENDED的技巧看analyzed层确认UnresolvedAttribute是否全部消失COUNT(*)的类型是否为LongType。若还有Unresolved*说明 Analyzer 失败查 Catalog 或字段名拼写看optimized层对比analyzed找Project、Filter节点是否被合并或下移。若WHERE还在顶层说明PushDownPredicate未生效看spark层重点看HashAggregate是否出现而非SortAggregateBroadcastHashJoin的右表是否是你期望的小表如dim_cityFileSourceScan的PartitionFilters是否包含你的dt条件。一个经典误判EXPLAIN显示BroadcastHashJoin但执行时却慢如蜗牛。原因往往是Broadcast的表实际大小远超spark.sql.autoBroadcastJoinThreshold。此时EXPLAIN COST会暴露真相BroadcastHashJoin节点的stats.sizeInBytes显示1.2GB而阈值是10MBSpark 实际执行时会 fallback 到SortMergeJoin但EXPLAIN没提示。解决方案调大阈值或强制/* SHUFFLE_HASH(t2) */。4.3 源码断点调试在 IntelliJ 中精准捕获 Catalyst 的心跳本地调试是理解源码的终极手段。步骤如下以 Spark 3.5.0 Scala 2.12 为例导入工程git clone https://github.com/apache/spark.git用 IntelliJ 打开选择Import project from external model → SBT配置 Run ConfigurationAdd Configuration → ApplicationMain class 设为org.apache.spark.sql.SparkSessionVM options 加-Dspark.sql.adaptive.enabledfalse避免 AQE 干扰设置断点Parser在AstBuilder.visitSelect()第一行Analyzer在ResolveRelations.apply()的catalog.lookupRelation调用前Optimizer在PushDownPredicate.apply()的case j: Join 分支Planner在JoinSelection.apply()的if (canBroadcast(right))判断处Executor在HashAggregateExec.doExecute()的new HashAggregateIterator(childIter)行运行并观察执行spark.sql(SELECT ...)断点停住后用Evaluate Expression查看plan、ctx、conf等变量Step Into追踪调用栈。注意Catalyst 大量使用 Scala 的PartialFunction和transformIntelliJ 的Step Over有时会跳过内部逻辑。此时务必用Step Into并关注transform的f: TreeNode TreeNode参数它才是规则的实际执行体。另外TreeNode的transformUp自底向上和transformDown自顶向下行为不同调试时注意区分。4.4 常见问题速查表从现象、根因到修复的完整闭环现象可能根因定位方法修复方案验证方式AnalysisException: cannot resolve xxx given input columnsAnalyzer未找到字段可能是表别名未定义、字段名拼写错误、或USING子句中ON条件引用了不存在的列EXPLAIN EXTENDED查analyzed层看UnresolvedAttribute是否残留检查 SQL 中所有字段是否在FROM子句的表/别名中存在用DESCRIBE table_name确认 schemaEXPLAIN EXTENDED中analyzed层无Unresolved*查询极慢EXPLAIN显示SortMergeJoin但右表很小Planner误判右表大小spark.sql.autoBroadcastJoinThreshold过小或 Catalog 无统计信息EXPLAIN COST查BroadcastHashJoin节点的sizeInBytesDESCRIBE DETAIL table_name查statistics调大spark.sql.autoBroadcastJoinThreshold或ANALYZE TABLE table_name COMPUTE STATISTICSEXPLAIN中BroadcastHashJoin出现且EXPLAIN COST显示大小合理OutOfMemoryError在HashAggregateExec数据倾斜导致单个 partition 的哈希表过大或spark.sql.adaptive.enabledfalse无法动态调整内存spark.sql.adaptive.logLevelINFO查 AQE 日志jstack看线程堆栈是否卡在HashAggregateIterator.next()开启spark.sql.adaptive.enabledtrue或加/* REPARTITION(key) */强制重分区AQE 日志显示SkewJoinOptimization或CoalesceShufflePartitions被触发EXPLAIN显示计划正常但结果为空或错误Executor层序列化失败如 UDF 返回null未处理、或InternalRow字段顺序与 schema 不匹配查driver和executor日志中的SerializationExceptionspark.sql.adaptive.enabledfalse下用EXPLAIN对比optimized和spark层UDF 中增加null检查确保Row.fromSeq的字段顺序与StructType严格一致用spark.range(10).map(...)小数据集复现并调试 UDFPushDownPredicate未下推WHERE条件condition中包含非确定性函数如rand(),current_date()或Catalog未提供TableStatsEXPLAIN EXTENDED查optimized层看Filter节点位置DESCRIBE FORMATTED table_name查Statistics避免在WHERE中用非确定性函数或用/* PUSH_DOWN_FILTER(...) */Hint需自定义EXPLAIN EXTENDED中Filter节点出现在FileSourceScan下方最后分享一个小技巧当你怀疑某个规则失效但EXPLAIN又看不出端倪时可以临时在Optimizer.scala中defaultBatches里把该规则单独拎出来放在一个新 Batch 的第一行并加一行println(sApplying my rule on $plan)。重新编译 Spark用这个版本跑 SQL控制台会直接打印规则是否被调用、输入计划是什么。这是最直接的“白盒验证”法比猜强一百倍。5. 性能调优与高级实践从配置参数到自定义规则的深度掌控5.1 关键配置参数的“影响半径”与调优阈值Spark SQL 的性能