Iceberg 在微信实验平台的应用实践

摘要：导读在当今数字化时代，实验平台对于数据分析和决策起着至关重要的作用。微信实验平台处理着海量的数据，面临着高计算量和复杂的数据管理挑战。Iceberg 作为一种先进的数据存储架构，为解决这些问题提供了新的思路和方法。本文详细阐述了微信实验平台如何利用 Icebe

导读在当今数字化时代，实验平台对于数据分析和决策起着至关重要的作用。微信实验平台处理着海量的数据，面临着高计算量和复杂的数据管理挑战。Iceberg 作为一种先进的数据存储架构，为解决这些问题提供了新的思路和方法。本文详细阐述了微信实验平台如何利用 Iceberg 的特性，在多种场景下的优化实践。

主要内容包括以下几大部分：

1. 微信实验平台计算存储架构简介

2. Iceberg 湖仓改造和探索

3. 总结和收益

4. 后续规划

分享嘉宾｜郑森烈、王龙腾讯微信专家工程师

编辑整理｜王红雨

内容校对｜李瑶

出品社区｜DataFun

微信实验平台计算存储架构简介

1. 实验指标计算示例

以一个开了两天的实验为例，设有两张表，一张用于表征用户（UIN）是否命中，另一张记录业务方相关指标数值，如转发数、停留时长等。两天的指标计算需在每天内分别进行左右表的 join 以算出当天数据，若要计算两天内的平均效果，如人均转发次数，则需进行两天的 full join 并最终做聚合操作。实际业务中，存在数万个实验和指标，平均可能查看七到十五天的结果，因此每天会产生超过百 PB 的海量计算。

2. 平台计算存储架构概述

在微信实验平台的系统结构中，红色箭头代表数据的流向。命中数据来自后台 server 模块或微信客户端的上报，微信用户在业务方也会形成指标或画像数据。这两部分数据经预处理后，全部存储在 Iceberg 表中，为上层的 Spark 或 OLAP 引擎提供数据支持，计算结果用于实验结果分析。指标计算系统是从海量数据到实验结论的桥梁，目前微信实验平台每天有十万亿级的去重后命中记录，指标接了几千张业务表，每天 IO 超百 PB，计算消耗的 CPU 达每天百万核时。

Iceberg 湖仓改造和探索

1. 迁移 Iceberg 对大规模指标计算的优化

当前平台内部管理的命中表已经全量从 Hive 迁移到了 Iceberg。由于 Iceberg 元数据管理设计的优势，在相同分区量时，对 NameNode（NN）和 MetaStore 的压力更小。在十万亿级数据量上，实验平台使用每天万级的分区量。业务方的表管理现状是，部分历史存量表在 Hive 上，新接表基本都采用 Iceberg。

在业务表中可能会存在一些大的热点表，例如某直播业务表，每天有 3T 数据量且关联两千个指标，这意味着每天需对该表进行两千次计算。通过对指标定义上的公共字段做统计并进行分区转写，每天 IO 和 open 文件的量降低了 80%。利用 Iceberg 做数据裁剪，可以显著降低计算的 IO 量。

同时，借助 Iceberg Merge Into 的特性，平台由日度命中数据生成了累计命中数据。此外，还会利用 time travel 和分支 tag 功能，实现历史数据的存储查询。

2. 基于 Iceberg-API 实现湖上的 no shuffle join

Parquet 作为 Iceberg 的一种文件存储格式，内部由 Row Group 做数据分段，每个分段内有 min-max 索引，使用中对文件内数据排序可大幅提升 IO 效率。我们的测试表明，对数据进行概念上分桶，点查分桶内及一定范围内分桶的数据，速度可实现数量级的提升。在微信实验的压缩域存储和计算场景中，我们把默认的 128M 数据分段改为 4M（通过参数控制）。如果对在压缩域上存储和计算感兴趣，可以参考我们发表的论文：Large-Scale Metric Computation in Online Controlled Experiment Platform，VLDB 2024。

实现湖上 no shuffle join 的另一个基础是 Iceberg 对外开放了丰富的 API，可用于数据拉取、表更新和事务操作等。

no shuffle join 的原理如图所示，在 Spark 任务内，不同 executor 可直接拉取左右表对应的数据到本地，然后做本地的 local join，左右表数据可在同一文件或不同文件。

实际使用中，我们会广播 Iceberg 表元信息和 PlanFile 的结果，用来复用对 MetaStore 的访问和表元文件解析。

不同 no shuffle join 方案各有其优缺点。在 HDFS 上直接维护索引也可实现 no shuffle，但操作麻烦且不兼容上层查询。在 Hive 和 Iceberg 中，如 Iceberg 的 SPJ（ Storage Partitioned Join），使用分桶表，对应分桶被同一台机器拉取做本地 join，对 SQL 的优化更加通用，但可能存在潜在的分桶小文件问题。我们的 no shuffle join 方案特点是灵活、可编程性强，没有小文件问题，但同时需要少量编程才可使用。线上的实际效果是，在一些指标任务中，可以把执行时间从五分钟优化到四十秒，一般原任务内 join 的次数越多，优化幅度越大。

3. 湖表合并提交优化

业务上的背景是，微信实验平台会识别不同用户配置的指标（如人均点赞数、停留时长、转发数等），如果这些指标来自同一个业务表，并具有类似的 where 和 group by 等条件，会自动合并转写成一条 SQL，这样可以一次性算出多个指标，复用 IO 和 shuffle 结果。

然而，业务表上配置的众多指标并非都能合并，部分因条件不同需放到不同 Spark 任务里执行。对于计算结果明细的存储，可以有多种方案：第 1 种方案是，来源于同一个业务表的明细结果也存到同一个结果表，这样图中并行的 Spark1-4 之间会存在写冲突；第 2 个方案，每个指标存一张表，它的缺点是需要对结果数据串行写和提交，效率较低；第 3 种方案是，每个任务存到随机的一张表里做负载均衡，这种方案的问题是查询复杂，使用时要先查表名再查数据，单次查某个指标的多天明细需要查多张表，对 MetaStore 的访问也会放大。以上方案在一次性补录多个日期时，不同天之间的任务也可能会发生提交冲突，这样会降低写表效率，严重时会导致任务失败。

实验平台的优化方案是，每个 Spark 任务直接将数据写入指标对应的分表，把得到的 Data File 注册到对应分表的等待队列，每次提交会合并队列里所有 Data File，实现跨任务的合并。性能测试表明，原始 Iceberg 方案并发大时性能退化严重，使用时先加分布式锁然后提交，效率会有所提升，而跨任务合并提交每分钟可成功提交两百次以上。同时，需注意不同任务写同一分区时可能产生数据冲突，需要进行仲裁，此合并方案也适用于其他高频 commit 的场景。

4. StarRocks/ ClickHouse 与湖仓的结合

我们在使用 ClickHouse 进行实时和离线指标计算时，发现不同机器节点的线程同时读表会瞬间给 Iceberg 表带来较大请求量，增加元数据压力。微信的 OLAP 团队对此进行优化，不同节点可以复用对同一张表查询的元数据。此外，由于 ClickHouse 分布式 join 能力不完备，我们采用了 Co-locate Join 的方式计算指标，类似 Spark no shuffle join，每台机器仅拉取对应的逻辑分桶数据到本地进行 join。同时，由于 ClickHouse 使用的 SSD 容量有限且成本较高，我们在使用中将其作为缓存，按需从 Iceberg 加载数据做计算，并采用 LRU 策略做空间回收。

微信实验平台在 StarRocks 上也有相关应用，基于 Iceberg 和 StarRocks 在上层做了分布式的因果推断。在 StarRocks 里实现了与假设检验相关的算子（函数），用户可用一条 SQL 运行较复杂的模型进行检验，实现各种分析，相关工作也回馈给了社区。如果对分布式因果推断感兴趣，可以参考我们的开源项目：https://github.com/Tencent/fast-causal-inference

总结和收益

1. 存储和计算成本降低

通过利用 Iceberg 多级分区、Merge Into、Time Travel 等特性进行优化后，平台的存储和计算成本每年降低千万级别。

2. 计算效率提升

基于湖表跨任务合并 commit 等优化，计算效率得到了显著提升，指标结果的实时性更好，在补录以及数据恢复情况下速度更快，可靠性增加。

3. 实现 no shuffle join

在 Iceberg 开放的能力上实现了 no shuffle join，支持分钟级到秒级的实验明细导出。

4. 支持指标计算与分析

利用 ClickHouse/StarRocks 支持指标计算，并在此基础上建立了分布式因果推断分析。

后续规划

1. 更简洁的流批一体建设

当前我们支持 10min 级/小时级/天极离线三类指标计算，在数据接入、存储和计算的流批一体建设上还有继续提升的空间。