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

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

---

先说结论

快手 AB 场景的核心指标模板，迁移前后的对比：

指标	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 → 未生效，需排查：

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

---

快手的 Production 数据验证

快手生产环境中，Colocate Join 的实际效果：

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

也就是说：大部分数据在本地节点内就被消化掉了，网络中只传输聚合后的少量结果集

---

进阶：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 所属实验组）。

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

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

---

总结：Checklist

如果你也要在 Doris 上做 AB 指标计算或类似场景，按这个顺序排查：

• 建表时按 Join Key 哈希分桶
• 两张表分桶数 严格一致
• 加入同一个 colocate_with Group
• EXPLAIN 验证出现 CollocatedJoin
• 大基数 GROUPING SETS + DISTINCT 启用 Local Distinct
• CPU 热路径 UDF 改为 C++ Native UDF

---

如果觉得有帮助，欢迎点赞关注 有问题可以在评论区交流～

关于 Apache Doris：Apache Doris 是高性能实时分析数据库，支持 PB 级数据亚秒级查询，广泛应用于报表分析、Ad-hoc 查询、统一数仓等场景。SelectDB 是 Apache Doris 的商业化公司，提供企业级支持和云服务。欢迎加入 Doris 社区 交流更多实践。