分布式计算有一个朴素的承诺：数据量翻倍，机器也翻倍，耗时就不变。只要数据分得均匀，性能问题大多可以靠加资源解决。可现实中总有这样的作业：加了一倍资源，耗时纹丝不动。原因通常只有两个：

1. 数据分得不均：大部分数据挤在少数几个任务上，其他任务早早算完在等它，这就是数据倾斜；

2. 数据搬得太多：为了把相同 key 的数据凑到一起，全量数据要在节点之间重新分发，这就是 Shuffle。

这篇文章就按这两条线索，从使用者的视角看怎么调优。

一、数据倾斜：木桶由最短的那块板决定

一个 Stage 要等所有任务都完成才算完成。如果按 key 分配任务时，某个 key 的数据特别多，它所在的任务就会比其他任务慢很多倍，整个作业的耗时就等于这个最慢任务的耗时，加再多机器也没用。

下面的实验模拟了这个过程：有一部分数据都是同一个热点 key，它们只能落到同一个任务上。试着调大「热点数据比例」，看看负载会变成什么样；再调多「任务数量」，看看能不能救回来。

可以看到，热点比例一高，增加任务数只会让其他任务更闲，那个最忙的任务一点没变轻。所以解决倾斜的思路只有一个：把热点 key 的数据拆开，分给更多任务。

先确认是不是倾斜：看作业监控里各任务的处理条数或耗时，个别任务远超其他任务就是倾斜；再用一条简单的统计找出热点 key：

SELECT user_id, COUNT(1) AS cnt
FROM orders
GROUP BY user_id
ORDER BY cnt DESC
LIMIT 20;

下面用两张表举例：订单表 orders 和用户表 users。

CREATE TABLE orders (
    order_id BIGINT COMMENT '订单ID',
    user_id  BIGINT COMMENT '用户ID',
    amount   BIGINT COMMENT '订单金额',
    region   STRING COMMENT '所在地区'
);

CREATE TABLE users (
    user_id   BIGINT COMMENT '用户ID',
    user_name STRING COMMENT '用户名称'
);

关联时的倾斜

需求是给订单补上用户名：

SELECT o.order_id, o.user_id, u.user_name, o.amount
FROM orders o
LEFT JOIN users u ON o.user_id = u.user_id;

Join 会把两边 user_id 相同的数据发到同一个任务上，所以倾斜来自某些 user_id 的订单特别多。分两种情况处理。

情况一：热点是无意义的值

最常见的是大量订单的 user_id 为空。空值本来就关联不上任何用户，没有必要把它们都送到同一个任务。可以把空值替换成随机的负数，既保证关联不上，又让它们均匀散开：

SELECT o.order_id, o.user_id, u.user_name, o.amount
FROM orders o
LEFT JOIN users u
-- 空值变成随机负数：关联不上，而且分散到各个任务
ON IF(o.user_id IS NULL, -1 - FLOOR(1024 * RAND()), o.user_id) = u.user_id;

情况二：热点是真实的大客户

如果 user_id IN (10001, 10002) 两个大客户占了大部分订单，就不能丢掉它们了。思路是加盐：把大表里热点 key 的每条数据随机打上 0~1023 的后缀，让它们分散到 1024 个任务；小表里同一个 key 则复制 1024 份，每份带一个后缀，保证无论分到哪个任务都能关联上。

-- 第一步：大表中热点 key 随机加后缀
WITH salted_orders AS (
    SELECT
        CASE WHEN user_id IN (10001, 10002)
             THEN CONCAT(user_id, '_', CAST(FLOOR(RAND() * 1024) AS STRING))
             ELSE CAST(user_id AS STRING)
        END AS salted_user_id,
        order_id, user_id, amount
    FROM orders
),
-- 第二步：小表中热点 key 复制 1024 份，每份带一个后缀
salted_users AS (
    SELECT
        CASE WHEN user_id IN (10001, 10002)
             THEN CONCAT(user_id, '_', CAST(n AS STRING))
             ELSE CAST(user_id AS STRING)
        END AS salted_user_id,
        user_name
    FROM users
    LATERAL VIEW EXPLODE(
        CASE WHEN user_id IN (10001, 10002)
             THEN SEQUENCE(0, 1023)  -- 热点 key 复制 1024 份
             ELSE ARRAY(0)           -- 其他 key 只保留 1 份
        END
    ) t AS n
)
-- 第三步：按加盐后的 key 关联
SELECT o.order_id, o.user_id, u.user_name, o.amount
FROM salted_orders o
LEFT JOIN salted_users u ON o.salted_user_id = u.salted_user_id;

聚合时的倾斜

需求换成按地区统计订单金额和下单人数：

SELECT
    region,
    COUNT(DISTINCT user_id) AS uv,
    SUM(amount) AS total_amount
FROM orders
GROUP BY region;

如果某个地区的订单特别多，它的所有数据都会涌向一个任务。解决办法是两阶段聚合：先按「地区 + 用户 ID 的哈希桶」做一次局部聚合，把一个大组拆成 1024 个小组；再按地区把局部结果加起来。

-- 第一阶段：按 地区 + 分桶 局部聚合
WITH agg_partial AS (
    SELECT
        region,
        COUNT(DISTINCT user_id) AS partial_uv,
        SUM(amount) AS partial_sum
    FROM orders
    GROUP BY region, MOD(HASH_CODE(user_id), 1024)
)
-- 第二阶段：按地区汇总
SELECT
    region,
    -- 同一个 user_id 只会落进同一个桶，所以各桶的 UV 可以直接相加
    SUM(partial_uv) AS uv,
    SUM(partial_sum) AS total_amount
FROM agg_partial
GROUP BY region;

SUM 可以拆开再加，这很自然；COUNT(DISTINCT) 也能相加，靠的是按用户 ID 分桶：同一个用户只在一个桶里被计数，桶与桶之间没有重复。为什么去重计数通常不能相加、什么时候又可以，大数据去重统计一文有更详细的讨论。

流计算里的思路相同。Flink SQL 的 Local-Global 聚合会在数据发往下游之前先在上游做一次局部聚合，相同 key 的数据先在本地合并，热点 key 发往下游的数据量就小多了：

引擎自带的开关

上面的改写手法要动 SQL，不少引擎已经内置了自动处理：

先试试这些开关，确实解决不了再手工改写。

二、少搬数据：MPP 引擎里的 Join

在 Doris 这样的 MPP 引擎里，Join 的主要成本往往不是计算，而是为了把相同 key 的数据凑到一起而发生的网络传输。所以优化 Join 的核心是：能不搬就不搬，必须搬就少搬。

选对分发方式

按需要搬动的数据量从多到少，常见的四种方式是：

Shuffle Join：两边都搬

最通用的方式：左右两张表都按关联键重新分发。两张表都很大、又没有其他条件可用时只能这样。

Broadcast Join：只搬小表

右表很小时，把整张小表复制给每个节点，大表原地不动。小表越小越划算；小表一旦变大，复制的成本会超过 Shuffle。

Bucket Shuffle Join：只搬一边

如果关联条件包含左表的分桶列，左表的数据本来就按这个键分好了桶，只需要把右表按左表的分桶规则发过去。把大表放在左边，就可以避免搬动大表。

Colocate Join：都不搬

如果两张表在建表时就按同样的列、同样的桶数分桶，并放在同一个 Colocation Group 里，相同 key 的数据本来就在同一个节点上，Join 可以完全在本地完成。代价是建表时就要规划好。

Runtime Filter：先过滤，再搬

大表关联小表时，往往只有一小部分大表数据能关联上。Runtime Filter 会在执行时根据小表里实际出现的关联键，生成一个过滤条件（比如 IN 列表、Bloom Filter 或最大最小值），提前下推到大表的扫描节点。关联不上的数据在扫描时就被过滤掉，根本不会进入网络。

总结

物理学家 P. W. Anderson 在《More Is Different》里说，量变到一定程度就会产生质变。数据处理也一样：数据量小的时候怎么写都行，数据量大到一定程度，就必须了解数据怎么分布、引擎怎么执行、数据怎么在节点之间流动，才能把性能调好。

但最有效的优化永远是少处理数据：只读需要的列和分区，过滤条件尽量早，能预先聚合就预先聚合。数据少了，倾斜和 Shuffle 的问题也会跟着变小。

参考资料

• P. W. Anderson，More Is Different，Science 177(4047)，1972。

• Spark SQL Performance Tuning：自适应执行与倾斜 Join 优化。

• Flink Performance Tuning：MiniBatch、Local-Global 聚合与 Split Distinct。

• Apache Doris 文档：Join 方式与 Runtime Filter 相关章节。

• xkcd #435: Purity