我们在使用不同的引擎进行大数据计算时，需要将数据根据计算引擎进行适配。这是一个相当棘手的问题，为此出现了一种新的解决方案：介于上层计算引擎和底层存储格式之间的一个中间层。这个中间层不是数据存储的方式，只是定义了数据的元数据组织方式，并向计算引擎提供统一的类似传统数据库中 “表” 的语义。它的底层仍然是 Parquet、ORC 等存储格式。

基于此，Netflix 开发了 Iceberg，目前已经是 Apache 的顶级项目，https://iceberg.apache.org/。

1.数据湖的解决方案 - Iceberg

1.1 Iceberg 是什么

Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to compute engines including Flink Trino Spark and Hive using a high-performance table format that works just like a SQL table.

Iceberg 是一种开放的数据湖表格式。可以简单理解为是基于计算层（Flink、Spark）和存储层（ORC，Parquet，Avro）的一个中间层，用 Flink 或者 Spark 将数据写入 Iceberg，然后再通过其他方式来读取这个表，比如 Spark，Flink，Presto 等。

在文件 Format（Parquet / Avro / ORC 等）之上实现 Table 语义：

• 支持定义和变更 Schema
• 支持 Hidden Partition 和 Partition 变更
• ACID 语义
• 历史版本回溯
• 借助 Partition 和 Columns 统计信息实现分区裁剪
• 不绑定任何存储引擎，可拓展到 HDFS / S3 / OSS 等
• 容许多个 writer 并发写入，乐观锁机制解决冲突

1.2 Iceberg 的 Table Format 介绍

Iceberg 是为分析海量数据而设计的，被定义为 Table Format，Table Format 介于计算层和存储层之间。

Table Format 向下管理在存储系统上的文件，向上为计算层提供丰富的接口。存储系统上的文件存储都会采用一定的组织形式，譬如读一张 Hive 表的时候，HDFS 文件系统会带一些 Partition、数据存储格式、数据压缩格式、数据存储 HDFS 目录的信息等，这些信息都存在 Metastore 上，Metastore 就可以称之为一种文件组织格式。

一个优秀的 文件组织格式，如 Iceberg，可以更高效的支持上层的计算层访问磁盘上的文件，做一些 list、rename 或者查找等操作。

表和表格式是两个概念。表是一个具象的概念，应用层面的概念，我们天天说的表是简单的行和列的组合。而 表格式 是数据库系统实现层面一个抽象的概念，它定义了一个表的 Scheme 定义：包含哪些字段，表下面文件的组织形式（Partition 方式）、元数据信息（表相关的统计信息，表索引信息以及表的读写 API），如下图左侧所示：

上图右侧是 Iceberg 在数据仓库生态中的位置，和它差不多相当的一个组件是 Metastore。不过 Metastore 是一个服务，而 Iceberg 就是一系列 jar 包。对于 Table Format，我认为主要包含 $4$ 个层面的含义，分别是 表 Schema 定义（是否支持复杂数据类型），表中文件的组织形式，表相关统计信息、表索引信息以及表的读写 API 信息。

• 表 Schema 定义了一个表支持字段类型，比如 int、string、long 以及复杂数据类型等。
• 表中文件组织形式最典型的是 Partition 模式，是 Range Partition 还是 Hash Partition。
• Metadata 数据统计信息。
• 表的读写 API。上层引擎通过对应的 API 读取或者写入表中的数据。

1.3 Iceberg 的核心思想

Iceberg 的核心思想，就是 在时间轴上跟踪表的所有变化：

• 快照 表示表数据文件的一个完整集合。
• 每次更新操作会生成一个新的快照。

1.4 Iceberg 的元数据管理

从图中可以看到 Iceberg 将数据进行分层管理，主要分为 元数据管理层 和 数据存储层。元数据管理层又可以细分为三层：

• Metadata File
• Snapshot
• Manifest

Metadata File 存储当前版本的元数据信息（所有 Snapshot 信息）；Snapshot 表示当前操作的一个快照，每次 commit 都会生成一个快照，一个快照中包含多个 Manifest。每个 Manifest 中记录了当前操作生成数据所对应的文件地址，也就是 data files 的地址。基于 Snapshot 的管理方式，Iceberg 能够进行 time travel（历史版本读取以及增量读取），并且提供了 serializable isolation。

数据存储层支持不同的文件格式，目前支持 Parquet、ORC、AVRO。

1.5 Iceberg 的重要特性

Apache Iceberg 设计初衷是 为了解决 Hive 离线数仓计算慢的问题，经过多年迭代已经发展成为构建数据湖服务的表格式标准。关于 Apache Iceberg 的更多介绍，请参见 Apache Iceberg 官网。

目前 Iceberg 提供以下核心能力：

1.5.1 丰富的计算引擎

• 优秀的内核抽象使之不绑定特定引擎，目前在支持的有 Spark、Flink、Presto、Hive。
• Iceberg 提供了 Java Native API，不用特定引擎也可以访问 Iceberg 表。

1.5.2 灵活的文件组织形式

• 提供了 基于流式的增量计算模型 和 基于批处理的全量表计算模型，批任务和流任务可以使用相同的存储模型（HDFS、OZONE），数据不再孤立，以构建低成本的轻量级数据湖存储服务。
• Iceberg 支持隐藏分区（Hidden Partitioning）和分区布局变更（Partition Evolution），方便业务进行数据分区策略更新。
• 支持 Parquet、ORC、Avro 等存储格式。

1.5.3 优化数据入湖流程

• Iceberg 提供 ACID 事务能力，上游数据写入即可见，不影响当前数据处理任务，这大大简化了 ETL。
• Iceberg 提供 Upsert / Merge Into 行级别数据变更，可以极大地缩小数据入库延迟。

1.5.4 增量读取处理能力

• Iceberg 支持通过流式方式读取增量数据，实现主流开源计算引擎入湖和分析场景的完善对接。
• 支持 Spark Structed Streaming。
• 支持 Flink Table Source。
• 支持历史版本回溯。

1.6 数据文件结构

我们先了解一下 Iceberg 在文件系统中的布局，总体来讲 Iceberg 分为两部分数据。

• 第一部分是 数据文件，如下图中的 .parquet 文件。
• 第二部分是 表元数据文件（Metadata 文件），包含 Snapshot 文件（snap-*.avro）、Manifest 文件（.avro）、TableMetadata 文件（*.json）等。

1.6.1 元数据文件

其中 Metadata 目录存放元数据管理层的数据，表的元数据是不可修改的，并且始终向前迭代；当前的快照可以回退。

1.6.1.1 Table Metadata

version[number].metadata.json：存储每个版本的数据更改项。

1.6.1.2 快照（Snapshot）

snap-[snapshotID]-[attemptID]-[commitUUID].avro：存储快照 Snapshot 文件。

快照代表一张 Iceberg 表在某一时刻的状态，也被称为 清单列表（Manifest List），里面存储的是清单文件列表，每个清单文件占用一行数据。清单列表文件以 snap 开头，以 avro 后缀结尾，每次更新都产生一个清单列表文件。每行中存储了清单文件的路径。

清单文件（Manifest Files）里面存储数据文件的分区范围、增加了几个数据文件、删除了几个数据文件等信息。数据文件（Data Files）存储在不同的 Manifest Files 里面，Manifest Files 存储在一个 Manifest List 文件里面，而一个 Manifest List 文件代表一个快照。

1.6.1.3 清单文件（Manifest File）

[commitUUID]-[attemptID]-[manifestCount].avro：Manifest 文件。

清单文件是以 avro 格式进行存储的，以 avro 后缀结尾，每次更新操作都会产生多个清单文件。其里面列出了组成某个快照（Snapshot）的数据文件列表。每行都是每个数据文件的详细描述，包括 数据文件的状态、文件路径、分区信息、列级别的统计信息（比如每列的最大最小值、空值数等）、文件的大小 以及 文件里面数据的行数 等信息。其中列级别的统计信息在 Scan 的时候可以为算子下推提供数据，以便可以过滤掉不必要的文件。

1.6.2 数据文件

data 目录组织形式类似于 Hive，都是以分区进行目录组织（图中 dt 为分区列）。

Iceberg 的数据文件通常存放在 data 目录下。一共有三种存储格式（Avro、ORC 和 Parquet），主要是看你选择哪种存储格式，后缀分别对应 avro、orc 或者 parquet。在一个目录，通常会产生多个数据文件。

2.Apache Iceberg 的实现细节

2.1 快照设计方式

2.1.1 快照隔离

• 读操作仅适用当前已生成快照。
• 写操作会生成新的隔离快照，并在写完成后原子性提交。

如下图所示，虚线框（Snapshot-1）表示正在进行写操作，但是还没有发生 commit 操作，这时候 Snapshot-1 是不可读的，用户只能读取已经 commit 之后的 Snapshot。同理，Snapshot-2，Snapshot-3 表示已经可读。

可以支持并发读，例如可以同时读取 S1、S2、S3 的快照数据，同时，可以回溯到 Snapshot-2 或者 Snapshot-3。在 Snapshot-4 commit 完成之后，这时候 Snapshot-4 已经变成实线，就可以读取数据了。

例如，现在 Current Snapshot 的指针移到 S3，用户对一张表的读操作，都是读 Current Snapshot 指针所指向的 Snapshot，但不会影响前面的 Snapshot 的读操作。

当一切准备完毕之后，会以原子操作的方式 commit 这个 Metadata 文件，这样一次 Iceberg 的数据写入就完成了。随着每次的写入，Iceberg 就生成了下图这样的一个文件组织模式。

2.1.2 增量读取数据

Iceberg 的每个 Snapshot 都包含前一个 Snapshot 的所有数据，每次都相当于全量读取数据，对于整个链路来说，读取数据的代价是非常高的。

如果我们只想读取当前时刻的增量数据，就可以根据 Iceberg 中 Snapshot 的回溯机制来实现，仅读取 Snapshot-1 到 Snapshot-2 的增量数据，也就是下图中的紫色数据部分。

同理，S3 也可以只读取红色部分的增量数据，也可以读取 S1 - S3 的增量数据。

Iceberg 支持读写分离，也就是说可以支持并发读和增量读。

2.1.3 原子性操作

对于文件列表的所有修改都是原子操作。

• 在分区中追加数据。
• 合并或是重写分区。

• Iceberg 是以 文件 为粒度提交事务的，所以就没有办法做到以秒为单位提交事务，否则会造成文件数据量膨胀。
• 比如 Flink 是以 CheckPoint 为写入单位，物理数据在写入 Iceberg 之后并不能被直接查询，只有当触发了 CheckPoint 时才会写 Metadata，这时数据才会由不可见变成可见。而每次 CheckPoint 执行也需要一定的时间。

2.2 事务性提交

2.2.1 写操作要求

原子性替换保证了线性的历史。原子性替换需要依靠以下操作来保证：

• 记录当前元数据的版本 base version。
• 创建新的元数据以及 Manifest 文件。
• 原子性的将 base version 替换为新的版本。

2.2.2 冲突解决 - 乐观锁

• 假定当前没有其他的写操作。
• 遇到冲突则基于当前最新的元数据进行重试。
• 元数据管理器所提供的能力。
• HDFS 或是本地文件系统所提供的原子化的 rename 能力。

3.Iceberg 结合 Flink 场景分享

3.1 构建近实时 Data Pipeline

Iceberg 可以做到分钟级别的准实时数据拉取。

首先，Flink Iceberg 最经典的一个场景就是 构建实时的 Data Pipeline。业务端产生的大量日志数据，被导入到 Kafka 这样的消息队列。运用 Flink 流计算引擎执行 ETL 后，导入到 Apache Iceberg 原始表中。有一些业务场景需要直接跑分析作业来分析原始表的数据，而另外一些业务需要对数据做进一步的提纯。那么我们可以再新起一个 Flink 作业从 Apache Iceberg 表中消费增量数据，经过处理之后写入到提纯之后的 Iceberg 表中。此时，可能还有业务需要对数据做进一步的聚合，那么我们继续在 Iceberg 表上启动增量 Flink 作业，将聚合之后的数据结果写入到聚合表中。

有人会想，这个场景好像通过 Flink Hive 也能实现。 Flink Hive 的确可以实现，但写入到 Hive 的数据更多地是为了实现数仓的数据分析，而不是为了做增量拉取。一般来说，Hive 的增量写入以 Partition 为单位，时间是 $15min$ 以上，Flink 长期高频率地写入会造成 Partition 膨胀。而 Iceberg 容许实现 $1min$ 甚至 $30s$ 的增量写入，这样就可以大大提高了端到端数据的实时性，上层的分析作业可以看到更新的数据，下游的增量作业可以读取到更新的数据。

3.2 CDC 数据实时摄入摄出

Flink CDC（Change Data Capture）增量数据写入 Iceberg。

• 支持准实时的数据入湖和数据分析。
• 计算引擎原生支持 CDC，无需添加额外的组件。
• 采用统一的数据湖存储方案，并支持多种数据分析引擎。
• 支持增量数据读取。

可以用 Flink Iceberg 来分析来自 MySQL 等关系型数据库的 binlog 等。一方面，Apache Flink 已经原生地支持 CDC 数据解析，一条 binlog 数据通过 ververica flink-cdc-connector 拉取之后，自动转换成 Flink Runtime 能识别的 INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER 四种消息，供用户做进一步的实时计算。

此外，CDC 数据成功入湖 Iceberg 之后，我们还会打通常见的计算引擎，例如 Presto、Spark、Hive 等，他们都可以实时地读取到 Iceberg 表中的最新数据。

MySQL Binlog 是二进制格式的日志文件，但是不能把 binlog 文件等同于 OS 系统某目录下的具体文件，这是狭隘的。Binlog 是用来记录 MySQL 内部对数据库的改动（只记录对数据的修改操作），主要用于数据库的主从复制以及增量恢复。

3.3 从 Iceberg 历史数据启动 Flink 任务

上面的架构是采用 Iceberg 全量数据和 Kafka 的增量数据来驱动新的 Flink 作业。如果需要过去很长时间例如一年的数据，可以采用常见的 Lambda 架构，离线链路通过 Kafka → Flink → Iceberg 同步写入到数据湖，由于 Kafka 成本较高，保留最近 $7$ 天数据即可，Iceberg 存储成本较低，可以存储全量的历史数据，启动新 Flink 作业的时候，只需要去拉 Iceberg 的数据，跑完之后平滑地对接到 Kafka 数据即可。

3.4 通过 Iceberg 数据来修正实时聚合结果

同样是在 Lambda 架构下，实时链路由于事件丢失或者到达顺序的问题，可能导致流计算端结果不一定完全准确，这时候一般都需要全量的历史数据来订正实时计算的结果。而我们的 Iceberg 可以很好地充当这个角色，因为它可以高性价比地管理好历史数据。

4.Iceberg 0.11.1 源代码编译

4.1 编译 Iceberg

构建 Iceberg 需要 Grade $5.6$ 和 Java $8$ 的环境。

4.1.1 下载 Iceberg 0.11.1 软件包

下载地址：

• https://github.com/apache/iceberg/releases/tag/apache-iceberg-0.11.1
• https://www.apache.org/dyn/closer.cgi/iceberg/apache-iceberg-0.11.0/apache-iceberg-0.11.0.tar.gz

4.1.2 解压 Iceberg 0.11.1 软件包

[bigdata@bigdata185 software]$ tar -zxvf iceberg-apache-iceberg-0.11.1.tar.gz -C /opt/module/ 
[bigdata@bigdata185 software]$ cd /opt/module/iceberg-apache-iceberg-0.11.1/

4.1.3 修改对应的版本

我们选择最稳定的版本进行编译，Hadoop $\mathrm{2.7.7}$、Hive $\mathrm{2.3.9}$、Flink $\mathrm{1.11.6}$、Spark $\mathrm{3.0.3}$。

org.apache.flink:* = 1.11.6 
org.apache.hadoop:* = 2.7.7
org.apache.hive:hive-metastore = 2.3.9 
org.apache.hive:hive-serde = 2.3.9 
org.apache.spark:spark-hive_2.12 = 3.0.3

4.1.4 编辑 build.gradle 文件，添加国内源

（1）在 buildscript 的 repositories 中添加：

maven { url 'https://mirrors.huaweicloud.com/repository/maven/' }

添加后如下所示：

buildscript {
  repositories {
    jcenter()
    gradlePluginPortal()
    maven { url 'https://mirrors.huaweicloud.com/repository/maven/' }
  }
  dependencies {
    classpath 'com.github.jengelman.gradle.plugins:shadow:5.0.0'
    classpath 'com.palantir.baseline:gradle-baseline-java:3.36.2'
    classpath 'com.palantir.gradle.gitversion:gradle-git-version:0.12.3'
    classpath 'com.diffplug.spotless:spotless-plugin-gradle:3.14.0'
    classpath 'gradle.plugin.org.inferred:gradle-processors:2.1.0'
    classpath 'me.champeau.gradle:jmh-gradle-plugin:0.4.8'
  }
}

（2）allprojects 中添加：

maven { url 'https://mirrors.huaweicloud.com/repository/maven/' }

添加后如下所示

allprojects {
  group = "org.apache.iceberg"
  version = getProjectVersion()
  repositories {
    maven { url 'https://mirrors.huaweicloud.com/repository/maven/' }
    mavenCentral()
    mavenLocal()
  }
}

4.1.5 下载依赖（可选）

进入项目根目录，执行脚本：

[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew dependencies

4.1.6 正式编译

（1）进入项目根目录，执行：

[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew build

（2）上述命令会执行代码里的单元测试，如果不需要，则执行以下命令：

[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew build -x test -x scalaStyle

4.1.7 生成的目录

4.2 Iceberg 环境部署

在后面的章节中，我们分别介绍如何集成 Iceberg 0.11.1 和 Flink 1.11.6、Spark 3.0.3、Hive 2.3.9。

5.总结

• 数据湖的解决方案 Iceberg 介绍。
• Apache Iceberg 的技术实现细节。
• Iceberg 结合 Flink 场景分享。
• Iceberg $\mathrm{0.11.1}$ 源码编译。