首页 > 数据库 >Doris实战：Colocate Join消除Shuffle，AB指标计算性能提升145倍

Doris实战：Colocate Join消除Shuffle，AB指标计算性能提升145倍

来源：互联网 2026-07-08 08:29:05

在快手AB测试场景中，通过ApacheDoris的ColocateJoin消除跨节点Shuffle，单链路耗时从约21分钟降至秒级，性能提升145倍，资源消耗下降72%，且数据在本地完成Join，先过滤约95%无效数据，仅传输少量聚合结果。

关键词：Apache Doris · Colocate Join · AB 测试 · 性能优化 · SelectDB · 实时分析

长期稳定更新的攒劲资源： >>>点此立即查看<<<

先说结论

今天聊一个在快手AB测试场景中实战验证过的优化手段——Colocate Join。先看迁移前后的对比数据，直观感受一下差距：

指标	Spark	Apache Doris（优化后）
单链路耗时	~21 分钟	秒级（提速 145 倍）
资源消耗	基准	下降 72%
集群规模	-	2000 BE / 10 万核

这里面最关键的一步，就是Colocate Join 消除了跨节点 Shuffle。单个 Bucket 扫描千万级数据，本地先过滤掉约 95% 无效数据，最终只传输几千行结果。这种差距，不是简单的线性优化，而是量级上的跃升。

Colocate Join 到底是什么？

简单说，就是让相同 Join Key 的数据，在写入时就落到同一台机器的同一个分桶内。

这样查询时，两张表可以直接在本地完成 Join，不需要跨节点 Shuffle 搬运数据。数据在不跨节点的情况下直接完成本地关联计算。换句话说，数据搬家的过程被彻底省掉了。

没有 Colocate 时（Spark / 默认 Doris）

 复制代码Node A: [UID 1, 3, 5...]  ← Shuffle →  Node B: [UID 2, 4, 6...]
Node B: [UID 2, 4, 6...]  ← Shuffle →  Node A: [UID 1, 3, 5...]

网络传输量 = 全量数据。

有 Colocate 时

 复制代码Node A: [UID 1, 3, 5...]  ← 本地 Join →  结果在 Node A
Node B: [UID 2, 4, 6...]  ← 本地 Join →  结果在 Node B

网络传输量 = 聚合后的结果集（几千行）。

落地配置（完整可执行）

建表时满足两个条件，Colocate Join 自动生效。

条件 1：两张表按相同的 Join Key 分桶

 复制代码-- 累计分流表（记录用户命中哪个实验）
CREATE TABLE exp_bucket_table (
    uid BIGINT COMMENT '用户ID，也是 Join Key',
    experiment_id VARCHAR(128),
    group_id VARCHAR(64),
    bucket_id INT,
    ...
) ENGINE=OLAP
UNIQUE KEY(uid, experiment_id)
DISTRIBUTED BY HASH(uid) BUCKETS 128
PROPERTIES (
    "replication_num" = "3",
    "colocate_with" = "ab_colocate_group"  -- 关键：加入同一个 Colocate Group
);

 复制代码-- 指标宽表（记录用户行为指标）
CREATE TABLE metric_wide_table (
    uid BIGINT COMMENT '用户ID，也是 Join Key',
    metric_a DOUBLE,
    metric_b DOUBLE,
    dt DATE
) ENGINE=OLAP
DUPLICATE KEY(uid, dt)
DISTRIBUTED BY HASH(uid) BUCKETS 128  -- 关键：分桶数必须相同！
PROPERTIES (
    "replication_num" = "3",
    "colocate_with" = "ab_colocate_group"  -- 关键：同一个 Colocate Group
);

条件 2：分桶数严格一致

两张表的 BUCKETS 必须完全相同（上例均为 128），否则 Doris 无法保证数据分布对齐，Colocate Join 不会生效。

条件 3：验证是否生效

 复制代码-- 看执行计划，确认 Colocate Join 是否生效
EXPLAIN
SELECT
    e.experiment_id,
    e.group_id,
    COUNT(*) AS user_cnt,
    SUM(m.metric_a) AS sum_metric_a
FROM exp_bucket_table e
JOIN metric_wide_table m ON e.uid = m.uid
WHERE e.dt = '2024-01-01'
GROUP BY e.experiment_id, e.group_id;

执行计划中间出现 CollocatedJoin → 生效。

执行计划中仍有 Exchange / Shuffle → 未生效，需排查：

两张表的分桶数是否一致？
两张表是否加入了同一个 colocate_with Group？
Join 条件是否正好是分桶键（UID）？

快手的 Production 数据验证

在快手的生产环境中，Colocate Join 的实际效果非常直观：

单个 Bucket 扫描 千万级数据
本地 Join 先过滤掉约 95% 的无效数据
只保留 几十万行匹配结果
最终输出 几千行

换句话说：大部分数据在本地节点内就被消化掉了，网络中只传输聚合后的少量结果集。这才是Colocate Join的核心逻辑。

进阶：Local Distinct Grouping Sets

Colocate Join 消除了 Join 的 Shuffle，但 GROUPING SETS + DISTINCT 仍可能触发全局 Shuffle。

快手的做法是：把去重计算前移到各节点本地执行，先局部 Distinct，再对结果全局汇总。

 复制代码-- 透明改写模式（推荐）：优化器自动识别，SQL 无需改动
SELECT
    experiment_id,
    group_id,
    COUNT(DISTINCT uid),
    SUM(metric_a)
FROM exp_bucket_table e
JOIN metric_wide_table m ON e.uid = m.uid
GROUP BY experiment_id, group_id
GROUPING SETS ((experiment_id), (experiment_id, group_id));

注意：这个优化适合大基数 GROUPING SETS + DISTINCT（当 Shuffle 是瓶颈时很有效）；但如果是小基数的 COUNT DISTINCT，就不建议使用，因为会引入约 6 秒 Barrier 等待，得不偿失。

进阶：C++ Native UDF 压缩 CPU 热路径

AB 链路还有一个 CPU 热路径，就是分流判定 UDF（对每条日志判断 UID 所属实验组）。

快手把 Ja va UDF 改写为 C++ Native UDF，三项优化叠加下来，整体性能提升约 3 倍：

优化	原实现	优化后
P0 字符串拼接	每行创建 String 对象（~30% CPU）	ThreadLocal 复用 Buffer
P1 配置访问	每行 Unordered Map 查询	初始化展开为数组，O(1) 访问
P2 对象构造	每行构造智能指针	直接从 Block 列读原始值