字节跳动湖平台在批计算和特征场景的实践

DataFunTalk 2023-05-11

538

导读随着业务的发展，字节跳动特征存储已到达 EB 级别，日均增量 PB 级别，每天训练资源量级为百万 Core。随之而来的是内部业务方对原始数据存储、特征回填需求、降低成本、提升速度等需求的期待。本次分享将围绕问题背景、选型 & Iceberg 简介、基于 Iceberg 的实践及未来规划展开。

全文目录如下：

1. 问题背景

2. 选型 & Iceberg 简介

3. 基于 Iceberg 的实践

4. 未来规划

分享嘉宾｜刘纬字节跳动火山引擎云原生计算研发工程师

编辑整理｜王吉东昆仑数据

出品社区｜DataFun

问题背景

1. 用户使用流程

如我们所知，字节跳动是一家擅长做 A/B test 的公司。以特征工程调研场景为例，流程如下：

（1）首先由算法工程师进行在线特征抽取。

（2）将抽取到的特征，使用 Protobuf 的格式按行存至 HDFS。

出于存储成本的考量，一般只存储抽取后的特征，而不存储原始特征。

（3）将 HDFS 存储的特征交由字节自研的分布式框架（ Primus ）进行并发读取，并进行编码和解码操作，进而发送给训练器。

（4）由训练器对模型进行高效训练。

① 如果模型训练效果符合算法工程师的预期，说明该调研特征生效，进而算法工程师对调研特征进行回溯，通过 Spark 作业将特征回填到历史数据中，分享给其他算法工程师，进而迭代更多的优质模型。

② 如果模型训练效果不符合算法工程师的预期，则调研特征不对原有特征集合产生影响。

2. 业务规模

公司庞大的业务规模，带来了巨大的计算和存储体量：

（1）特征存储总量达 EB 级。

（2）单表特征最大可达百 PB 级（如广告业务）。

（3）单日特征存储增量达 PB 级。

（4）单日训练资源开销达 PB 级。

3. 遇到的问题

当特征调研场景叠加巨大的数据体量，将会遇到以下困难：

（1）特征存储空间占用较大。

（2）样本读放大，不能列裁剪，很难落特征进样本。

（3）样本写放大，COW 很难做特征回溯调研。

（4）不支持特征 Schema 校验。

（5）平台端到端体验差，用户使用成本高。

选型 & Iceberg 简介

在特征调研场景下，行存储是个低效的存储方式；因此，我们选择 Iceberg 存储方式来解决上述问题。

1. 整体分层

Apache Iceberg 是由 Netflix 公司推出的一种用于大型分析表的高性能通用表格式实现方案。

如上图所示，系统分成引擎层、表格式层、文件格式层、缓存加速层、对象存储层。图中可以看出，Iceberg 所处的层级和 Hudi，DeltaLake 等工具一样，都是表格式层。

（1）向上提供统一的操作 API

Iceberg 定义表元数据信息以及 API 接口，包括表字段信息、表文件组织形式、表索引信息、表统计信息以及上层查询引擎读取、表写入文件接口等，使得 Spark， Flink 等计算引擎能够同时高效使用相同的表。

（2）下层有 parquet、orc、avro 等文件格式可供选择。

（3）下接缓存加速层，包括开源的 Alluxio、字节火山引擎自研的 CFS 等。

CFS 全称是 Cloud File System，是面向火山引擎和专有云场景下的大数据统一存储服务，支持高性能的缓存和带宽加速，提供兼容 HDFS API 的访问接口。

（4）最底层的实际物理存储，可以选择对象存储，比如 AWS S3，字节火山引擎的 TOS，或者可以直接使用 HDFS。

通过上图可以比较清晰地了解到，Iceberg 这个抽象层最大的优势在于：将底层文件的细节对用户屏蔽，将上层的计算与下层的存储进行分离，从而在存储和计算的选择上更为灵活，用户可以通过表的方式去访问，无需关心底层文件的信息。

2. Iceberg 简介

（1）Iceberg 架构

Iceberg 的本质是一种文件的组织形式。如上图所示，包括多级的结构：

① Iceberg Catalog：用于保存表和存储路径的映射关系，其核心信息是保存 Version 文件所在的目录。

Iceberg Catalog 共有8种实现方式，包括 HadoopCatalog，HiveCatalog，JDBCCatalog，RestCatalog 等
不同的实现方式，其底层存储信息会略有不同；RestCatalog 方式无需对接任何一种具体的存储，而是通过提供 Restful API 接口，借助 Web 服务实现 Catalog，进一步实现了底层存储的解耦。

② Metadata File：用来存储表的元数据信息，包括表的 schema、分区信息、快照信息（ Snapshot ）等。

Snapshot 是快照信息，表示表在某一时刻的状态；用户每次对 Table 进行一次写操作，均会生成一个新的 SnapShot。
Manifestlist 是清单文件列表，用于存储单个快照的清单文件。
Manifestfile 是存储的每个数据文件对应的清单文件，用来追踪这个数据文件的位置、分区信息、列的最大最小值、是否存在 null 值等统计信息。

③ Data File 是存储的数据，数据将以 Parquet、Orc、Avro 等文件格式进行存储。

（2）Iceberg 特点

① SchemaEvolution：Iceberg 表结构的更新，本质是内在元信息的更新，因此无需进行数据迁移或数据重写。Iceberg 保证模式的演化（Schema Evolution）是个独立的、没有副作用的操作流程，不会涉及到重写数据文件等操作。

② Time travel：用户可任意读取历史时刻的相关数据，并使用完全相同的快照进行重复查询。

③ MVCC：Iceberg 通过 MVCC 来支持事务，解决读写冲突的问题。

④ 开放标准：Iceberg 不绑定任何计算引擎，拥有完全独立开放的标准，易于拓展。

（3）Iceberg 读写流程和提交流程

① 读写

每次 Iceberg 的写操作，只有发生 commit 之后，才是可读的；如有多个线程同时在读，但一部分线程在写，可以实现：只有 commit 完整的数据之后，对用户的读操作才能被用户的读线程所看到，实现读写分离。
例如上图中，在对 S3 进行写操作的时候，S2、S1 的读操作是不受影响的；此时 S3 无法被读到，只有 commit 之后 S3 才会被读到。此时 Current Snapshot 会指向 S3。
Iceberg 默认从最新 Current Snapshot 读取数据；如果读更早的数据，可指定对应的Snapshot 的 id ，实现数据回溯。

② 事务性提交

写操作：记录当前元数据的版本——base version，创建新的元数据以及 manifest 文件，原子性将 base version 替换为新的版本。
原子性替换：原子性替换保证了线性历史，通过元数据管理器所提供的能力，以及 HDFS 或本地文件系统所提供的原子化 rename 能力实现。
冲突解决：基于乐观锁实现，每一个 writer 假定当前没有其他的写操作，对表的 write 进行原子性的 commit，若遇到冲突则基于当前最新的元数据进行重试。

（4）分区裁剪

① 直接定位到 parquet 文件，无需调用文件系统的 list 操作。

② Partition 的存储方式对用户透明，用户在修改 partition 定义时，Iceberg 可以自动地修改存储布局，无需用户重复操作。

（5）谓词下推

Iceberg 会在两个层面实现谓词下推：

① 在 snapshot 层面，过滤掉不满足条件的 data file。

② 在 data file 层面，过滤掉不满足条件的数据。

其中，snapshot 层面的过滤操作为 Iceberg 所特有，正是利用到 manifest 文件中的元数据信息，逐字段实现文件的筛选，大大地减少了文件的扫描量。而同为 Table Format 产品、在字节其他业务产线已投入使用的 Hudi，虽然同样具备分区剪枝功能，但是尚不具备谓词下推功能。

基于 Iceberg 的实践

Hudi、Iceberg、DeltaLake 这三款 TableFormat 产品各有优劣，然而并没有任何一款产品能够直接满足我们的使用场景需求；考虑到 Iceberg 具备良好的 Schema Evolution 能力，支持下推，且无需绑定计算引擎等优点，因此字节选择使用 Iceberg 作为数据湖工具。

1. 整体架构

① 在字节的整体架构中，最上层是业务层，包含抖音，头条，小说等字节绝大部分业务线，以及火山引擎云原生计算等相关 ToB 产品（如 Seveless Spark 等）。

② 在平台层，使用 Global Lake Service 给业务方提供简单易用的 UI 和访问控制等功能。

③ 在框架层，我们使用 Spark 作为特征处理框架（包含预处理和特征调研等），使用字节自研的 Primus 分布式框架作为训练框架，使用 Flink 实现流式训练。

④ 在格式层，选择 parquet 作为文件格式，使用 Iceberg 作为表格式。

⑤ 最下层是调度器层和存储层。选择 Yarn 和 K8S 作为调度器；存储层一般选择 HDFS 进行存储，对于 ToB 产品，则使用 CFS 进行存储。

2. Data-Parquet

结合上图可以看出，列存储在特征调研场景存在以下优势：

① 可选择指定列进行读取：有效减少读放大问题，同时易于增列，即新增一列的时候，只需单独写入一列即可，元数据信息会记录每一列所在的磁盘位置。

② 压缩：同一列的数据格式相同，因此具有更好的压缩比；同一列的数据名称相同，因此无需进行冗余字符串存储。

③ 谓词下推：对每一列数据记录相应的统计信息（如 min，max 等），因此可以实现部分的谓词下推。

为了解决业务方的痛点问题，我们改成使用 Parquet 列存储格式，以降低数据的存储成本；同时由于 Parquet 选列具备下推到存储层的特性，在训练时只需读取模型所需要的特征即可，从而降低训练时序列化、反序列化的成本，提升训练的速度。

然而使用 Parquet 列存储，带来优点的同时也相应地带来了一些问题：

① 原来的行存储方式是基于 Protobuf 定义的半结构化数据，无需预先定义 Schema；然而使用 Parquet 之后，需要预先指定 Schema 才能进行数据的存取；这样在特征新增和淘汰的时候，Schema 的更新将会变成一个棘手的问题。

② 此外，Parquet 不支持数据回填；如果需要要回填比较长的数据，就需要将数据全量读取，增加新列，再全量写回。这样一方面会造成大量计算资源的浪费，另一方面会带来 overwrite 操作，导致正在进行训练的任务由于文件被替换而失败。

为了解决以上两个问题，我们引入了 Iceberg 来支持 SchemaEvolution，特征回填以及并发读写。

3. 特征回填

（1）COW

从上图可以看出，使用 Iceberg COW（Copy on write）方式进行特征回填，通过 BackFill 任务将原快照中的数据全部读出来，然后添加新列，进而写出到新的 Data File 中，并生成新的快照。

这种方式的缺点在于，仅仅新增一列数据的写入，却需要整体数据全部读出随后再全部写回，不仅浪费了大量的计算资源和存储资源；因此，我们基于开源的 Iceberg 自研了一种 MOR（Merge on Read）的 BackFill 方案。

（2）MOR

从上图可以看出，在 MOR 方案中，我们仍然需要一个 BackFill 任务来读取原始的 Data File 文件；所不同的是，我们只需读取少数需要的字段。比如我们需要对 A 列通过一些计算逻辑生成 C 列，那么 BackFill 任务只需从 Snapshot1 中读取 A 列的数据，且只需将 C 列的 update 文件写入 Snapshot2。

随着新增列的增多，我们需要将 update 文件合并到 Data File 里面；为此，我们进一步提供一种 Compaction 逻辑，即读取旧的 Data File 和 Update File，并合并生成新的 Data File。实现细节如下：

① 旧 Data File 和 Update File 都需要一个主键，并且每个文件都需要按照主键排序。

② 读取旧 Data File 时，会根据用户选择的列，分析具体需要哪些 Update File 和 Data File。

③ 根据旧 Data File 中 min-max 值去选择对应的 Update File。

由此可以看出，MOR 的本质是对多个 Data File 文件和 Update File 文件进行多路归并，而归并的顺序由 SEQ 决定，SEQ 大的数据（表明数据越新）会覆盖 SEQ 小的数据。

（3）两种特征回填方式对比

① COW

读写放大严重。
存储空间浪费。
读取逻辑简单。
写入耗费更多资源。
读取无需额外计算资源。

② MOR

没有读写放大。
节省存储空间。
读取逻辑复杂。
写入耗费较少资源。
绝大多数场景，不需要额外资源。

相比于 COW 方式的全量读取和写入，MOR 的优势在于只读取需要的列，同样也只写入更新的列，因此避免了读写放大的问题，节省大量计算资源，并大大降低读写 I/O；相比 COW 方式每次 COW 都翻倍的情况，MOR 只需存储新增列，大量节省了存储资源。

对于模型训练任务而言，大多数模型训练只需要用到少量的列，因此大量的线上模型都无需 MOR 操作，涉及开销可忽略不计；对于少数的特征调研模型，只需读取模型对应的 Update File 即可，因此带来的读取资源增加也非常有限。

4. 其他

除了上面提到的借助 Compaction 提高读性能以及分析特征删除场景外，我们还提供了以下几个服务：

（1）Expiration

① Snapshot Expiration：用于处理过期的 Snapshots。过期 Snapshots 不及时清理，会导致元数据文件堆积，从而带来文件膨胀问题，会给算法工程师带来困扰，因此需要服务定期做一些清理。我们通过平台化改造实现 Snapshots 文件的统一维护和清理；

② Data Expiration：大部分数据是有新鲜度和时效性的，因此用户可设置数据保存多久后被清理。

（2）CleanUp

由于一些事务的失败，或者一些快照的过期，导致文件在元数据文件中已经不再被引用，需要定期清理掉。

（3）Roll-Back

对于一些在 table 中非预期数据或者 schema 变更，希望将其回滚到之前稳定的 snapshot；结合平台的事件管理器，可以比较容易的实现这一功能。

（4） Statistics

用来实现一些湖平台可视化信息的展示，以及后端服务给业务带来的价值归纳。

5. 平台化改造

这里分享下字节内部实现的平台化工作。上图是批式特征存储的列表，借助站内实现的湖平台化工作，业务部门可以轻松实现特征的可视化操作，以及信息概览的获取。

下图是一张特征表样例，通过这张表可以直观地看到存储空间的使用、文件数的统计、记录数统计、特征统计等信息。这张表并不是数据规模最大的表，最大的表来自抖广，百 PB 量级的数据，千万级别的文件数。

未来规划

1. 规划重点

未来规划方面，我们计划逐步支持以下功能：

（1）湖冷热分层

在成本优化方面，我们规划做湖冷热分层。前文提到对于保存超过一定时间的数据，可以直接删除；然而对于一些场景，这些数据还会被使用，只是访问频率较低；因此未来考虑增加数据湖冷热分层功能，帮助用户降低成本。

（2）物化视图

在查询优化方面，我们计划做物化视图。这个也是源于我们 ToB 客户的真实需求，用来提升查询性能。目前这部分优化工作正处于一个商业化交付的流程中，大家也可以后续在我们的产品上进行体验。

（3）Self-Optimize

在体验优化方面，实现 Self-Optimize，例如前文提到的一些数据维护优化等。

（4）支持更多引擎

在生态丰富方面，我们会支持更多的引擎对接 Iceberg。

2. 整体平台架构总览

如前文所述，该平台不仅支持公司内部的业务，还会支持一定的 ToB 的业务；这些在字节内部实现的功能，以及未来我们规划的能力也会基于内外一致的思路去做演进；最终都会落到上图中涉及到的几款云原生计算产品上，如流式计算 Flink 版，云原生消息引擎 BMQ，云搜索服务 OpenSearch，大数据文件存储 CloudFS 等。目前该产品尚未转正式收费，感兴趣的朋友可以登录火山引擎官网https://www.volcengine.com体验。

整体平台架构以计算引擎产品为核心，包含两部分服务：

（1）云原生管理控制

① Quota 服务。

② 租户管理服务。

③ 运行时管理。

④ 生态整合服务。

⑤ 交付部署服务。

⑥ 网关服务。

（2）云原生运维平台

① 组件服务生命周期管理。

② Helm Chart 管理。

③ 日志、审计。

④ 监控报警。

⑤ 容灾、高可用。

以上功能均为 Serverless 的全托管产品，让用户更聚焦于自己的业务逻辑，减少数据运维带来的困扰。

想了解更多的云原生大数据相关领域的技术，或者有一些问题想做更多的交流，可以关注下我们的微信公众号。

问答环节

Q1：数据湖应该具备哪些能力？

A1：其实当下开源的数据湖都不能完全满足我们的需求。我们对数据湖的要求是，具备高吞吐 I/O 侧读写能力，支持开放的 table format 标准，不绑定固定引擎，具备灵活的元数据管理能力，具备数据裁剪能力，以上是对当前开源数据湖具备的能力的整合。这部分，我们最终是选择 Iceberg 工具并进行增强处理，我们内部称之为猛犸湖。

Q2：为什么字节选择 Iceberg 而不是 Hudi？如何看待未来数据湖的发展？

A2：在特征调研这一特定场景下，所有开源 table format 工具都不能直接满足该场景的需求。在如此大的体量和规模下，解决样本的读写方法问题，这是我们的出发点。相比于 Hudi，Iceberg 支持谓词下推功能，并且具备更强大的元数据管理能力，支持 schema evolution，并具备开放标准等优点，因此在这一特定场景下选择使用 Iceberg 而不是 Hudi。从未来数据湖的发展方向来看，我们基于 Iceberg 自研并扩展了一系列功能特性在特征调研场景下进行大规模实践，事实证明确确实实实现了大量存储成本的降低，以及特征调研链路的优化，因此这基于 Iceberg 扩展这样的数据湖技术发展方向仍然会有不错的前景。

今天的分享就到这里，谢谢大家。