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

先说结论

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

这里面最关键的一步，就是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% 的无效数据
• 只保留 几十万行匹配结果
• 最终输出 几千行

换句话说：大部分数据在本地节点内就被消化掉了，网络中只传输聚合后的少量结果集。这才是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 倍：

总结：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 的商业化公司，提供企业级支持和云服务。