Catalyst 优化器是 Spark SQL 的核心查询优化引擎，负责将用户写的 SQL 查询或 DataFrame 操作转换为高效的执行计划。

Catalyst 优化器的核心定义

Catalyst 是一个基于规则和成本的查询优化框架，采用函数式编程思想构建，具有以下特点：

• 可扩展的优化器架构
• 基于树结构的查询表示
• 多阶段优化流程
• 支持自定义优化规则

核心架构组成

四个主要阶段

内部组件详解

1. Parser（解析器）

// 将 SQL 字符串解析为抽象语法树 (AST)
val sqlText = "SELECT name, age FROM users WHERE age > 25"
val ast = spark.sessionState.sqlParser.parsePlan(sqlText)
// 输出: Project([name#0, age#1], Filter((age#1 > 25)), Relation(users))

2. Analyzer（分析器）

// 解析表名、列名，绑定元数据
val unresolvedPlan = parser.parsePlan("SELECT name FROM users")
val resolvedPlan = analyzer.execute(unresolvedPlan)
// 验证 users 表存在，name 列存在，并分配正确的数据类型

3. Logical Optimizer（逻辑优化器）

// 应用各种优化规则
val optimizerRules = Seq(
  // 谓词下推：将过滤条件下推到数据源
  PushDownPredicate,
  // 列裁剪：只读取需要的列
  ColumnPruning,
  // 常量折叠：预先计算常量表达式
  ConstantFolding,
  // 联接重排序：优化联接顺序
  ReorderJoin
)

4. Physical Planner（物理计划器）

// 选择最优的物理执行策略
val strategies = Seq(
  HashAggregation,
  SortAggregation,
  BroadcastHashJoin,
  ShuffledHashJoin,
  SortMergeJoin
)

主要优化技术

1. 谓词下推（Predicate Pushdown）

// 优化前的逻辑计划
val df = spark.read.parquet("data/sales.parquet")
val result = df.filter($"date" >= "2024-01-01").select("product", "amount")

// Catalyst 优化后会将过滤条件下推到 Parquet Reader
// 实际只会读取满足条件的分区数据，大幅减少 IO

2. 列裁剪（Column Pruning）

// 原始查询涉及大量列
val expensiveQuery = spark.sql("""
  SELECT user_id, COUNT(*) 
  FROM very_wide_table 
  GROUP BY user_id
""")

// Catalyst 自动识别只需要 user_id 列，忽略其他99个列
// 减少内存使用和网络传输开销

3. 常量折叠（Constant Folding）

// 用户写法
val df = spark.sql("SELECT *, salary * 1.1 + 1000 as adjusted_salary FROM employees")

// Catalyst 优化
// 如果某些值在编译时就能确定，则提前计算
// 例如：1.1 * 1000 = 1100，在运行时直接使用结果

4. Join 优化策略

// 广播 Join 优化
val smallDF = spark.range(1000).toDF("id")
val largeDF = spark.range(1000000).toDF("id")

// 当小表足够小时，自动选择广播 Join
val joined = smallDF.join(largeDF, "id")
// Catalyst 会选择 BroadcastHashJoin 策略

// Join 重排序优化
val complexJoin = df1.join(df2, "key1").join(df3, "key2").join(df4, "key3")
// Catalyst 会根据表大小重新安排 Join 顺序以最小化 Shuffle 开销

执行计划可视化

查看逻辑计划

val df = spark.sql("SELECT department, AVG(salary) FROM employees GROUP BY department")

// 未优化的逻辑计划
println("原始逻辑计划:")
df.queryExecution.logical.foreach(plan => println(plan))

// 优化后的逻辑计划
println("优化后逻辑计划:")
df.queryExecution.optimizedPlan.foreach(plan => println(plan))

查看物理计划

// 物理计划（带成本估算）
println("物理计划:")
df.queryExecution.sparkPlan.foreach(plan => println(plan))

// 可执行计划
println("可执行计划:")
df.queryExecution.executedPlan.foreach(plan => println(plan))

// 完整执行计划说明
df.explain(true)  // 显示所有阶段的详细信息

自适应查询执行（AQE）

动态优化特性

// 启用 AQE
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

// 示例：动态分区合并
val skewedData = spark.sql("""
  SELECT /*+ REPARTITION(1000) */ *
  FROM large_table
  WHERE filter_condition
""")
// AQE 会检测到实际只有少量分区有数据，自动合并空分区

性能监控和调试

监控优化效果

// 获取详细的执行统计
val metrics = df.queryExecution.executedPlan.metrics
metrics.foreach { case (name, metric) =>
  println(s"$name: ${metric.value}")
}

// 查看是否应用了特定优化
def checkOptimizationApplied(df: DataFrame, optimizationName: String): Boolean = {
  val planString = df.queryExecution.toString()
  planString.contains(optimizationName)
}

// 检查谓词下推是否生效
val predicatePushedDown = checkOptimizationApplied(resultDF, "PushedFilters")

自定义优化规则

// 创建自定义优化规则
object CustomOptimizerRule extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
    case Filter(condition, child) if condition == Literal.True =>
      // 移除恒真过滤条件
      child
  }
}

// 注册自定义规则
spark.experimental.extraOptimizations = 
  spark.experimental.extraOptimizations :+ CustomOptimizerRule

与其他系统的对比优势

与传统查询优化器比较

性能提升示例

// 未经优化的查询可能需要扫描 1TB 数据
// 经过 Catalyst 优化后：
// - 谓词下推减少到 100GB
// - 列裁剪减少到 50GB  
// - 分区裁剪减少到 10GB
// 总体性能提升可达 100x+

Catalyst 优化器的价值在于它能够自动化地应用数十种优化技术，使开发者无需深入了解底层实现细节就能获得接近专家级的查询性能。这种智能化的优化能力是 Spark SQL 区别于传统大数据处理框架的重要优势之一。