Spark SQL 查询优化架构

Catalyst 优化器核心组件

Catalyst 优化器详解

1. 逻辑优化阶段

常量折叠（Constant Folding）：

// 优化前
val df = spark.range(1, 100).filter((id + 1) * 2 > 10)

// 优化后等价于
val df = spark.range(1, 100).filter(id * 2 + 2 > 10)

谓词下推（Predicate Pushdown）：

// 原始查询
val result = ordersDF
    .join(customersDF, "customer_id")
    .filter(col("order_date") > "2024-01-01")

// 优化后：先过滤再连接
val optimized = ordersDF
    .filter(col("order_date") > "2024-01-01")
    .join(customersDF, "customer_id")

列裁剪（Column Pruning）：

// 原始查询（读取所有列）
val df = spark.read.parquet("large_table.parquet")
    .select("id", "name")
    .filter(col("status") == "active")

// 优化后：只读取需要的列
// 实际只读取 id, name, status 三列

2. 物理优化阶段

广播连接优化（Broadcast Join）：

// 小表自动广播
val smallTable = spark.table("dim_customers")  // 1000行
val largeTable = spark.table("fact_orders")    // 1亿行

val result = largeTable.join(smallTable, "customer_id")
// Catalyst 自动将 smallTable 广播到所有 executor

排序合并连接 vs 广播连接：

// 根据数据大小自动选择连接策略
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

Explain 语句详解

Explain 的基本用法

// 创建测试数据
case class Sales(region: String, product: String, amount: Double, date: String)
val salesDF = Seq(
    Sales("North", "Laptop", 1500.0, "2024-01-15"),
    Sales("South", "Phone", 800.0, "2024-01-20")
).toDF()

val customersDF = Seq(
    ("North", "Enterprise"),
    ("South", "SMB")
).toDF("region", "segment")

// 复杂查询示例
val complexQuery = salesDF
    .join(customersDF, "region")
    .filter(col("amount") > 1000)
    .groupBy("segment")
    .agg(avg("amount").alias("avg_sale"))

// 使用 Explain
complexQuery.explain()

Explain 的不同模式

1. 简单模式（默认）

complexQuery.explain()

输出：

== Physical Plan ==
*(3) HashAggregate(keys=[segment#10], functions=[avg(amount#2)])
+- *(3) HashAggregate(keys=[segment#10], functions=[partial_avg(amount#2)])
   +- *(3) Project [segment#10, amount#2]
      +- *(3) BroadcastHashJoin [region#0], [region#8], Inner, BuildRight
         :- *(3) Filter (isnotnull(amount#2) && (amount#2 > 1000.0))
         :  +- *(3) LocalTableScan [region#0, amount#2]
         +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
            +- *(1) LocalTableScan [region#8, segment#10]

2. 扩展模式

complexQuery.explain("extended")

输出包含：

• 解析的逻辑计划
• 分析后的逻辑计划
• 优化后的逻辑计划
• 物理计划

3. 代码生成模式

complexQuery.explain("codegen")

显示生成的 Java 代码，用于性能调优。

4. 成本模式（需要统计信息）

// 先收集统计信息
salesDF.createOrReplaceTempView("sales")
spark.sql("ANALYZE TABLE sales COMPUTE STATISTICS")

complexQuery.explain("cost")

实际优化案例解析

案例 1：连接顺序优化

// 未优化的查询
val slowQuery = ordersDF
    .join(customersDF, "customer_id")        // 大表 join 大表
    .join(productsDF, "product_id")          // 结果集仍然很大
    .filter(col("order_date") > "2024-01-01")

slowQuery.explain()

优化建议：

// 优化后的查询
val fastQuery = ordersDF
    .filter(col("order_date") > "2024-01-01")  // 先过滤减少数据量
    .join(productsDF, "product_id")           // 连接较小的数据集
    .join(customersDF, "customer_id")

fastQuery.explain()

案例 2：避免数据倾斜

// 可能产生数据倾斜的查询
val skewedQuery = transactionsDF
    .groupBy("user_id")  // 某些用户可能有大量交易
    .agg(sum("amount").alias("total"))

skewedQuery.explain()

优化方案：

// 使用 salting 技术解决倾斜
import org.apache.spark.sql.functions._

val saltedDF = transactionsDF
    .withColumn("salt", (rand() * 10).cast("int"))
    .groupBy(col("user_id"), col("salt"))
    .agg(sum("amount").alias("partial_sum"))
    .groupBy("user_id")
    .agg(sum("partial_sum").alias("total"))

Explain 输出解读指南

关键操作符解读

扫描操作：

• LocalTableScan - 内存中的数据
• FileScan - 文件数据源
• InMemoryTableScan - 缓存的数据

连接操作：

• BroadcastHashJoin - 广播连接（最优）
• SortMergeJoin - 排序合并连接
• ShuffledHashJoin - 洗牌哈希连接

聚合操作：

• HashAggregate - 哈希聚合
• ObjectHashAggregate - 对象哈希聚合
• SortAggregate - 排序聚合

交换操作：

• Exchange - 数据洗牌
• BroadcastExchange - 广播交换

性能瓶颈识别

// 识别问题查询
val problematicQuery = largeDF
    .groupBy("category")
    .agg(collect_list("value").alias("values"))
    .filter(size(col("values")) > 100)

problematicQuery.explain()

常见问题模式：

1. 过多的 Exchange 操作 - 数据洗牌频繁
2. CartesianProduct - 笛卡尔积，性能杀手
3. 大量的 Sort 操作 - 排序开销大
4. BroadcastExchange 过大 - 广播数据量超标

高级优化技巧

1. 自定义优化规则

import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.plans.logical._

// 自定义优化规则示例
object CustomOptimizationRule extends Rule[LogicalPlan] {
    def apply(plan: LogicalPlan): LogicalPlan = plan transform {
        case Filter(condition, child) =>
            // 自定义过滤条件优化逻辑
            optimizeFilter(condition, child)
    }
    
    private def optimizeFilter(condition: Expression, child: LogicalPlan): LogicalPlan = {
        // 实现具体的优化逻辑
        Filter(condition, child)
    }
}

// 注册自定义规则
spark.experimental.extraOptimizations = Seq(CustomOptimizationRule)

2. 统计信息引导优化

// 收集表统计信息
spark.sql("ANALYZE TABLE sales COMPUTE STATISTICS FOR COLUMNS region, amount")

// 查看统计信息
spark.sql("DESCRIBE EXTENDED sales").show()

// 基于统计信息的优化会自动生效
val optimizedQuery = salesDF
    .filter(col("region") === "North")
    .join(customersDF, "region")

3. 自适应查询执行（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.skewedJoin.enabled", "true")

val aqeOptimized = largeDF
    .groupBy("key")
    .agg(sum("value").alias("total"))
    .join(otherDF, "key")

aqeOptimized.explain()

实战调试流程

性能问题排查步骤

def debugQueryPerformance(df: DataFrame): Unit = {
    println("=== 1. 简单执行计划 ===")
    df.explain()
    
    println("\n=== 2. 扩展执行计划 ===")
    df.explain("extended")
    
    println("\n=== 3. 检查数据分布 ===")
    df.foreachPartition { _ => 
        // 检查每个分区的数据量
    }
    
    println("\n=== 4. 检查存储格式 ===")
    // 确认使用列式存储（Parquet/ORC）
}

Explain 输出分析模板

def analyzeExplainPlan(plan: String): Unit = {
    val indicators = Map(
        "BroadcastHashJoin" -> "✅ 高效连接",
        "SortMergeJoin" -> "⚠️ 可能需优化", 
        "Exchange" -> "📊 数据洗牌",
        "CartesianProduct" -> "❌ 性能警告",
        "Filter" -> "🔍 过滤操作"
    )
    
    indicators.foreach { case (pattern, message) =>
        if (plan.contains(pattern)) println(s"$message: $pattern")
    }
}

总结

Explain 语句的核心价值

1. 查询理解 - 了解 Spark 如何执行你的查询
2. 性能诊断 - 识别瓶颈和优化机会
3. 优化验证 - 确认优化措施是否生效
4. 学习工具 - 深入理解 Catalyst 优化器工作原理

优化最佳实践

1. 始终查看 Explain 输出 - 在投入生产前检查执行计划
2. 优先使用内置优化 - 充分利用 Catalyst 的自动化优化
3. 合理设计数据模型 - 分区、分桶、统计信息收集
4. 适时使用手动优化 - 对于复杂场景进行针对性调优

通过掌握 Explain 语句和 Catalyst 优化器的工作原理，您可以编写出更高效、更可靠的 Spark SQL 查询。