Spark SQL Join优化梳理

在 Spark SQL 日常开发中，表关联（Join）是核心操作，也是性能调优的重灾区。不同关联场景（大表+小表、大表+大表）的优化思路差异显著，本文将分场景拆解最优优化方案，结合实操案例说明原理与落地方式。

一、大表关联小表优化（分2个子场景）

大表关联小表的核心优化目标是 减少/避免 Shuffle，优先触发高效的 BroadcastHashJoin，其次优化 SortMergeJoin 效率。

子场景1：小表可广播（体积 ≤ 50M）

核心思路

将小表全量广播到所有 Executor 节点，大表无需 Shuffle，直接在 Executor 本地完成 Join，是最优解。

1. 自动广播（默认策略，零侵入）

• 原理：Spark 内置参数 spark.sql.autoBroadcastJoinThreshold（默认10M）控制自动广播阈值，小表体积≤阈值时，自动触发 BroadcastHashJoin。
• 调优操作：

  -- 调整阈值为50M（Executor内存充足时）
  SET spark.sql.autoBroadcastJoinThreshold = 52428800;
  

• 注意：阈值不超过 Executor 内存的1/3，避免 OOM。

2. 手动强制广播（精准控制）

• 原理：通过 /*+ BROADCAST(小表名) */ HINT 强制广播，无视自动阈值，适用于小表体积10-50M场景。
• 实操示例：

  SELECT /*+ BROADCAST(dim_table) */
         fct.id, fct.value, dim.info
  FROM big_fact fct
  JOIN small_dim dim ON fct.dim_id = dim.id;
  

3. 小表预处理（缩容后广播）

• 原理：过滤无效数据、去重、裁剪冗余字段，缩小小表体积，满足广播条件。
• 实操示例：

  WITH filtered_dim AS (
      SELECT DISTINCT id, info  -- 去重+字段裁剪
      FROM small_dim
      WHERE info IS NOT NULL    -- 过滤无效数据
  )
  SELECT /*+ BROADCAST(filtered_dim) */
         fct.id, fct.value, filtered_dim.info
  FROM big_fact fct
  JOIN filtered_dim ON fct.dim_id = filtered_dim.id;
  

子场景2：小表不可广播（体积 > 50M）

核心思路

无法避免 Shuffle，需优化 SortMergeJoin 效率，重点解决数据倾斜、单分区数据量过大问题。

1. 基础优化：调整 Shuffle 分区数

• 原理：默认分区数 spark.sql.shuffle.partitions=200，单分区理想大小50-100M，按数据量调整分区数。
• 计算公式：目标分区数 =（大表数据量+小表数据量）/ 100M
• 实操示例：

  -- 大表50G+小表500M，设置分区数500
  SET spark.sql.shuffle.partitions = 500;
  

2. 进阶优化：加盐打散（解决热点 Key 倾斜）

• 原理：给小表 Join Key 加随机后缀（如0-9），大表扩容后按加盐 Key Join，打散热点 Key。
• 实操示例：

  -- 小表加盐
  WITH salted_dim AS (
      SELECT id, info, CAST(RAND()*10 AS INT) AS salt
      FROM medium_dim
  ),
  -- 大表扩容
  salted_fct AS (
      SELECT id, value, salt
      FROM big_fact
      LATERAL VIEW EXPLODE(ARRAY(0,1,2,3,4,5,6,7,8,9)) tmp AS salt
  )
  -- 加盐Join+去重
  SELECT fct.id, fct.value, dim.info
  FROM salted_fct fct
  JOIN salted_dim dim ON fct.id = dim.id AND fct.salt = dim.salt
  GROUP BY fct.id, fct.value, dim.info;
  

3. 批量优化：分桶 Join（频繁 Join 场景）

• 原理：小表按 Join Key 分桶，Spark 按桶直接 Join，减少 Shuffle 数据量。
• 实操步骤：

  -- 1. 创建分桶小表
  CREATE TABLE bucketed_dim (id INT, info STRING)
  CLUSTERED BY (id) INTO 20 BUCKETS
  STORED AS PARQUET;
  -- 2. 关闭自动广播，优先分桶Join
  SET spark.sql.autoBroadcastJoinThreshold = -1;
  SET spark.sql.join.preferSortMergeJoin = false;
  -- 3. 执行分桶Join
  SELECT fct.id, fct.value, dim.info
  FROM big_fact fct
  JOIN bucketed_dim dim ON fct.id = dim.id;
  

二、大表关联大表优化

大表关联大表无法避免大量 Shuffle，核心思路是 减少参与 Join 的数据量 + 优化数据分布 + 最大化利用集群资源。

1. 前置必做：数据过滤与字段裁剪

• 原理：Join 前过滤无效数据、裁剪冗余字段，减少 Shuffle 数据量（成本最低、效果最显著）。
• 实操示例：

  WITH filtered_fact AS (
      SELECT id, value  -- 只保留业务字段
      FROM big_fact1
      WHERE dt = '2024-01-01'  -- 分区过滤
  ),
  filtered_dim AS (
      SELECT id, info
      FROM big_fact2
      WHERE city = 'Hangzhou'  -- 行过滤
  )
  SELECT f.id, f.value, d.info
  FROM filtered_fact f
  JOIN filtered_dim d ON f.id = d.id;
  

2. 核心优化：分区裁剪 + 分区分桶 Join

• 原理：
  • 分区裁剪：只加载需要的分区数据，避免全表扫描；
  • 分桶 Join：按 Join Key 分桶，同桶数据在同一 Executor Join，减少 Shuffle。
• 实操示例：

  -- 两张大表均按dt分区、按id分桶，仅加载指定分区
  SELECT f.id, f.value, d.info
  FROM big_fact1 f
  JOIN big_fact2 d ON f.id = d.id
  WHERE f.dt = '2024-01-01' AND d.dt = '2024-01-01';
  

3. 进阶优化：加盐打散 + 热点 Key 单独处理

• 原理：对两张大表的 Join Key 同时加盐，打散热点 Key；对 NULL 值/极端热点 Key 单独过滤处理。
• 实操示例：

  -- 1. 过滤极端热点Key（NULL值）
  WITH fact_no_null AS (
      SELECT id, value FROM big_fact1 WHERE id IS NOT NULL
  ),
  dim_no_null AS (
      SELECT id, info FROM big_fact2 WHERE id IS NOT NULL
  ),
  -- 2. 两张大表加盐
  salted_fact AS (
      SELECT id, value, CAST(RAND()*10 AS INT) AS salt FROM fact_no_null
  ),
  salted_dim AS (
      SELECT id, info, CAST(RAND()*10 AS INT) AS salt FROM dim_no_null
  )
  -- 3. 加盐Join
  SELECT f.id, f.value, d.info
  FROM salted_fact f
  JOIN salted_dim d ON f.id = d.id AND f.salt = d.salt;
  

4. 终极优化：动态分区 Pruning（DPP）

• 原理：Spark 3.0+ 支持自动裁剪分区，无需手动指定分区条件，减少数据扫描量。
• 开启方式：

  SET spark.sql.optimizer.dynamicPartitionPruning.enabled = true;  -- 默认开启
  

• 实操示例：

  -- 自动裁剪big_fact2中city='Hangzhou'的分区
  SELECT f.id, f.value, d.info
  FROM big_fact1 f
  JOIN big_fact2 d ON f.id = d.id
  WHERE f.city = 'Hangzhou';
  

三、核心优化总结

关联场景	核心优化思路	最优方案
大表+小表（≤50M）	避免Shuffle，本地Join	自动/手动广播 + 小表预处理
大表+小表（>50M）	优化Shuffle，解决数据倾斜	调整分区数 + 加盐打散 + 分桶Join
大表+大表	减少数据量，优化数据分布	数据裁剪 + 分区分桶Join + 加盐打散 + DPP

四、调优验证要点

1. 查看 Spark UI → SQL 页面：优先出现 BroadcastHashJoin；
2. 查看 Stage 页面：Shuffle 读写数据量显著下降；
3. 查看 Task 页面：Task 执行时间均匀，无长尾 Task。

https://learn.lianglianglee.com/专栏/Spark性能调优实战/27 大表Join小表：广播变量容不下小表怎么办？.md