Tackling Data Skew in Large-Scale SQL Joins with MapJoin, DistMapJoin & SkewJoin

Scenario Description

Data skew is a classic interview topic for data engineers and occurs during join, group by, or count distinct operations when a small number of compute nodes receive a disproportionate amount of data. In Alibaba's payment data pipeline, large‑scale transaction tables are joined with merchant dimension tables, causing a few nodes to process far more records than others, which slows the entire job.

Common Optimization Methods

2.1 MapJoin

Broadcast a small dimension table to the compute nodes of the large fact table, eliminating the need for a shuffle on the large table. If the large table is already unevenly distributed, a random redistribution can be added before the MapJoin.

SELECT /*+MAPJOIN(dim)*/ * FROM (SELECT * FROM dwd_tbl) base LEFT OUTER JOIN (SELECT * FROM dim_tbl) dim ON base.dim_key = dim.dim_key;

2.2 Special/Null Value Dispersion

Replace empty or special placeholder values (e.g., "DEFAULT") with random values before the join to avoid concentrating records on a single node.

SELECT * FROM (SELECT * FROM dwd_tbl) base LEFT OUTER JOIN (SELECT * FROM dim_tbl) dim ON IF(COALESCE(base.dim_key,'' )='',CONCAT('HIVE_',RAND()),base.dim_key) = dim.dim_key;

2.3 Hotspot Dispersion with Multiplicative Expansion

Add a random integer field (1‑10) to the fact table and replicate the dimension table accordingly, then join on both the original key and the new fields.

SELECT * FROM (SELECT *,CAST(RAND()*10 AS BIGINT) AS ext_a FROM dwd_tbl) base LEFT OUTER JOIN (SELECT * FROM dim_tbl LATERAL VIEW EXPLODE(SPLIT('0;1;2;3;4;5;6;7;8;9',';')) tt AS ext_b) dim ON base.dim_key = dim.dim_key AND base.ext_a = dim.ext_b;

2.4 Hotspot‑Only Processing / SkewJoin

Extract hotspot keys, split the dimension table into hotspot and non‑hotspot parts, and process them separately: hotspot data with MapJoin, non‑hotspot data with a regular shuffle join, then union the results.

-- Extract hotspot keys
INSERT OVERWRITE TABLE tmp_hot_list PARTITION (dt='${bizdate}')
SELECT dim_shop_id AS hot_id FROM main_table WHERE dt='${bizdate}' GROUP BY dim_shop_id HAVING COUNT(1) > 10000;

-- Process hotspot data with MapJoin
SELECT /*+MAPJOIN(a2,a3)*/ a1.trade_no AS trade_no, a1.dim_shop_id AS shop_id, a3.shop_name, a3.shop_type
FROM (SELECT * FROM main_table WHERE dt='${bizdate}') a1
JOIN tmp_hot_list a2 ON a1.dim_shop_id = a2.dim_shop_id
LEFT OUTER JOIN (SELECT /*+MAPJOIN(b2)*/ b1.* FROM dim_table_info b1 JOIN tmp_hot_list b2 ON b1.dim_shop_id = b2.dim_shop_id) a3 ON a1.dim_shop_id = a3.dim_shop_id
UNION ALL
-- Process non‑hotspot data with normal join
SELECT /*+MAPJOIN(a12)*/ a11.trade_no AS trade_no, a11.dim_shop_id AS shop_id, a13.shop_name, a13.shop_type
FROM (SELECT * FROM main_table WHERE dt='${bizdate}') a11
LEFT ANTI JOIN tmp_hot_list a12 ON a11.dim_shop_id = a12.dim_shop_id
LEFT OUTER JOIN (SELECT /*+MAPJOIN(b12)*/ b11.* FROM dim_table_info b11 LEFT ANTI JOIN tmp_hot_list b12 ON b11.dim_shop_id = b12.dim_shop_id) a13 ON a11.dim_shop_id = a13.dim_shop_id;

2.5 Summary of Traditional Approaches

All methods aim to reduce the impact of shuffle‑induced hotspots, either by avoiding the shuffle (MapJoin) or by balancing data distribution during the shuffle.

New Idea: WithDistMapJoin

3.1 Core Concept

DistMapJoin builds remote distributed query nodes for medium‑size dimension tables, allowing the large fact table to join without any shuffle. This eliminates the primary cause of data skew while introducing a network‑level I/O hotspot that can be mitigated with query merging.

3.2 Implementation

WITH tmp_hot_pid AS (
    SELECT dim_shop_id, 'Y' AS is_hot FROM main_table_detail WHERE dt='${bizdate}' GROUP BY dim_shop_id HAVING COUNT(1) > 100000
),
tmp_dim_tbl AS (
    SELECT /*+MAPJOIN(hot)*/ dim.*, COALESCE(hot.is_hot,'N') AS is_hot FROM dim_table_info dim LEFT OUTER JOIN tmp_hot_pid hot ON dim.dim_shop_id = hot.dim_shop_id
),
tmp_dwd_tbl AS (
    SELECT /*+MAPJOIN(hot)*/ base.*, COALESCE(hot.is_hot,'N') AS is_hot FROM main_table_detail base LEFT OUTER JOIN tmp_hot_pid hot ON base.dim_shop_id = hot.dim_shop_id
)
INSERT OVERWRITE TABLE final_result_table PARTITION (dt='${bizdate}')
SELECT * FROM (
    -- Non‑hotspot data with DISTMAPJOIN
    SELECT /*+DISTMAPJOIN(dim(shard_count=77))*/ dwd_tbl.trade_no, dwd_tbl.trade_date, dwd_tbl.shop_id, dim.shop_name, dim.shop_type
    FROM (SELECT * FROM tmp_dwd_tbl WHERE is_hot='N') dwd_tbl
    LEFT OUTER JOIN (SELECT * FROM tmp_dim_tbl WHERE is_hot='N') dim ON dwd_tbl.partner_id = dim.partner_id
    UNION ALL
    -- Hotspot data with MAPJOIN
    SELECT /*+MAPJOIN(dim)*/ dwd_tbl.trade_no, dwd_tbl.trade_date, dwd_tbl.shop_id, dim.shop_name, dim.shop_type
    FROM (SELECT * FROM tmp_dwd_tbl WHERE is_hot='Y') dwd_tbl
    LEFT OUTER JOIN (SELECT * FROM tmp_dim_tbl WHERE is_hot='Y') dim ON dwd_tbl.partner_id = dim.partner_id
) base;

3.3 Real‑World Impact

Deployed in Alibaba's core payment data pipeline, the new scheme reduced average computation time by 40% and resource consumption by 30%, handling joins between transaction tables and merchant or contract dimension tables without any shuffle.

Overall Summary

The combined MapJoin + DistMapJoin solution effectively eliminates data skew caused by shuffle, offering superior performance for large‑scale SQL joins. For mild skew, a pure DistMapJoin may suffice, while more severe cases benefit from the hybrid approach.