Spark SQL 2.4 SinglePartition Exchange任务倾斜问题求助

看来你遇到的核心问题是两个：一是Window操作后的全局操作（比如LIMIT）触发了Exchange SinglePartition，导致所有数据被拉到单个节点处理；二是部分encnbr key数据量过大，引发数据倾斜。我来给你拆解可行的解决思路：

一、先定位问题根源 #

1. 确认是否存在超大encnbr key #

首先执行以下查询找出数据量最高的前10个encnbr，确认是否有某个key的行数远高于其他：

SELECT encnbr, COUNT(*) AS row_count
FROM (
    SELECT encnbr FROM table1 WHERE encnbr IS NOT NULL
    UNION ALL
    SELECT encnbr FROM table2 WHERE encnbr IS NOT NULL
) combined
GROUP BY encnbr
ORDER BY row_count DESC LIMIT 10;

2. 确认Exchange SinglePartition的来源 #

从你的执行计划看，Exchange SinglePartition在LocalLimit 4之上，说明你的实际查询末尾应该有全局LIMIT语句。全局LIMIT会强制Spark将所有数据shuffle到单个分区后取前N条，这是导致超长单任务的关键原因之一。

二、针对性解决方案 #

方案1：优化全局LIMIT，避免全量数据shuffle到单分区 #

如果问题源于全局LIMIT，我们可以先在每个分区提前做局部LIMIT，再全局筛选，减少单分区处理的数据量：

SELECT enc_key, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt, mod_flg
FROM (
    SELECT *,
           -- 全局排序生成行号
           ROW_NUMBER() OVER (ORDER BY eff_dt ASC, data_timestamp ASC) AS global_rn
    FROM (
        -- 原Window查询逻辑，保留排序字段方便后续全局排序
        SELECT enc_key, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt, 
               CASE 
                   WHEN lag(non_keys) OVER (PARTITION BY encnbr ORDER BY eff_dt ASC, data_timestamp ASC) IS NULL THEN 'Y'
                   WHEN lag(non_keys) <> non_keys THEN 'Y' 
                   ELSE 'N' 
               END AS mod_flg,
               eff_dt, data_timestamp
        FROM (
            SELECT enc_key, encnbr, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt, eff_dt, data_timestamp,
                   MD5(CONCAT(enc_key, prsn_key, prov_key, clm_key, clm_ln_key)) AS non_keys
            FROM table1 WHERE encnbr IS NOT NULL
            UNION ALL
            SELECT enc_key, encnbr, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt, eff_dt, data_timestamp,
                   MD5(CONCAT(enc_key, prsn_key, prov_key, clm_key, clm_ln_key)) AS non_keys
            FROM table2 WHERE encnbr IS NOT NULL
        ) base_data
        -- 每个分区先取足够多的条数（比如100条，远大于你需要的4条）
        LIMIT 100
    ) partition_limit_data
) global_limit_data
WHERE global_rn <= 4;

这样每个分区只输出100条数据，全局shuffle的数据量大幅减少，单分区任务不会再超长。

方案2：处理超大encnbr key引发的数据倾斜 #

如果查询中存在单个encnbr包含几十万甚至上百万条数据，我们可以拆分处理：

步骤1：分离大key与小key数据

WITH big_key_data AS (
    -- 替换成你找到的超大encnbr值
    SELECT enc_key, encnbr, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt, eff_dt, data_timestamp,
           MD5(CONCAT(enc_key, prsn_key, prov_key, clm_key, clm_ln_key)) AS non_keys
    FROM (
        SELECT * FROM table1 WHERE encnbr = '超大key值'
        UNION ALL
        SELECT * FROM table2 WHERE encnbr = '超大key值'
    ) t
),
small_key_data AS (
    SELECT enc_key, encnbr, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt, eff_dt, data_timestamp,
           MD5(CONCAT(enc_key, prsn_key, prov_key, clm_key, clm_ln_key)) AS non_keys
    FROM (
        SELECT * FROM table1 WHERE encnbr IS NOT NULL AND encnbr != '超大key值'
        UNION ALL
        SELECT * FROM table2 WHERE encnbr IS NOT NULL AND encnbr != '超大key值'
    ) t
)

步骤2：分别处理两类数据

• 大key数据：由于数据集中，直接在单分区内计算Window函数（无需shuffle多个分区）
• 小key数据：正常按encnbr分区处理

processed_big_key AS (
    SELECT enc_key, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt,
           CASE 
               WHEN lag(non_keys) OVER (ORDER BY eff_dt ASC, data_timestamp ASC) IS NULL THEN 'Y'
               WHEN lag(non_keys) <> non_keys THEN 'Y' 
               ELSE 'N' 
           END AS mod_flg
    FROM big_key_data
),
processed_small_key AS (
    SELECT enc_key, prsn_key, prov_key, clm_key, clm_ln_key, birth_dt,
           CASE 
               WHEN lag(non_keys) OVER (PARTITION BY encnbr ORDER BY eff_dt ASC, data_timestamp ASC) IS NULL THEN 'Y'
               WHEN lag(non_keys) <> non_keys THEN 'Y' 
               ELSE 'N' 
           END AS mod_flg
    FROM small_key_data
)
-- 合并结果
SELECT * FROM processed_big_key
UNION ALL
SELECT * FROM processed_small_key;

方案3：调整Spark参数辅助优化 #

• 增大shuffle分区数：设置spark.sql.shuffle.partitions=200（默认是20），让小key的数据更均匀分布在多个分区
• 开启倾斜优化：Spark 2.4支持数据倾斜自动优化，设置spark.sql.adaptive.enabled=true和spark.sql.adaptive.skewJoin.enabled=true，让Spark自动检测并处理倾斜情况

为什么CLUSTER BY encnbr没用？ #

CLUSTER BY只是保证相同encnbr的数据落在同一个分区并排序，但它无法解决单个encnbr数据量过大的问题——这个大key依然会占据整个分区，导致该分区任务超长。它更适合没有大key的场景，优化数据分布。

内容的提问来源于stack exchange，提问作者dexter80