导读： 今天分享的主题是Iceberg在腾讯的实践和优化。在这次分享中首先会介绍Iceberg，并深入Spark读写Iceberg的细节，结合具体案例展示我们在生产中遇到的问题、产生原因以及优化方案，最后会介绍我们围绕Iceberg做的数据治理服务。
全文目录：
Apache Iceberg介绍
Spark读写Iceberg
Iceberg生产实践
数据治理服务
分享嘉宾｜刘献杨 腾讯 高级工程师
编辑整理｜王羽凡 大疆
出品平台｜DataFunSummit
01
Apache Iceberg介绍
首先介绍一下Iceberg。
1. Apache Iceberg-表格式
Iceberg是一种开放的表格式，任何语言都可以遵循Iceberg社区定义的规范实现Iceberg。Apache Iceberg采用Java实现。正如它logo展示的一样， Iceberg可以管理海量的数据，但是向用户暴露的只是很小的一部分。
Iceberg有一些很好的特性：
支持事务 ：可以原子性的进行更新和删除数据
可扩展性强 ：meta data可以存在对象存储里，不像传统的Hive是存在数据库里
支持schema evolution和partition evolution
对存储层做了抽象 ：数据可以存在HDFS或者对象存储中，数据格式支持Parquet，ORC和Avro，支持混存，可以改变表的存储位置
2. Iceberg表的组成
Iceberg表由三部分组成， 上层是catalog层，中间是metadata层，下层是数据层。
Catalog层主要作为指针，指向表的存储位置。实现catalog需要提供原子性更新的操作，像Iceberg内置的HiveCatalog，HadoopCatalog，JDBCCatalog等都可以实现原子性的更新。
HiveCatalog是通过HMS的lock提供原子性的保证
HadoopCatalog是由HDFS rename提供原子性的保证
JDBCCatalog是通过数据库的lock来提供原子性的保证
Metadata层中最上层是JSON格式的metadata file，这个文件存有当前表的schema、partition等表的基础信息，以及很重要的ManifestList位置信息。ManifestList是Avro格式的文件，它由表在当前snapshot下所有的ManifestFile根据partition聚合而成，ManifestFile又是DataFile的聚合。
数据层中的DataFile记录有实际存储数据的文件（如Parquet文件）的地址以及其min-max和partition等文件元信息，用于文件过滤。
3. Iceberg表的ACID特性
Iceberg表的ACID特性是由Catalog来保证的，即更新操作在commit阶段需要持有Catalog锁，而在读写的过程中是不需要持有的。
以这张图为例，使用Spark做了某些原子性更新操作并commit，将表的版本由snapshot n-1转换到了snapshot n，后续由Flink做了一些其他操作，将snapshot n转换为snapshot n+1。这里的更新操作除了对表的增删改之外还包括了表格式的变更，比如改变了表的schema或者增加了table property。
Iceberg支持时间旅行，可以对历史的snapshot进行读取。因此Iceberg的数据读和数据写是可以并行的。
4. Iceberg Evolution
Iceberg对模式演变的支持很好。
①Schema Evolution
可以并行地修改表的schema，如增加/删除/修改列。
②Partition Evolution
与传统的Spark读Hive或者Parquet不同，Iceberg的partition是隐藏式的。比如创建一个表，这个表的partition是根据created_time这一列做transform得到的，用户读写表的时候不需要加上这个partition列。Partition列也可以被更新，比如原本根据月份做的partition，发现粒度不够细，可以再增加新的partition field以天作为更细粒度的partition，这些用户都是无感知的，并不需要改变写入和查询数据的sql语句。
--
02
Spark读写Iceberg
1. Spark写Iceberg表
Spark读写Iceberg是基于Apache Spark’s DataSourceV2 API实现的。数据读入会根据task plan生成多个WriteTask，每个WriteTask会写成一到多个DataFile。可以设置每个DataFile的target-file-size，比如设置成128M或者512M。每个DataFile记录的是实际存储的文件（如Parquet文件）的meta data信息，比如具体路径，min-max信息，partition信息。这些partition信息会聚合到drive端，生成一个由DataFile组成的列表。DataFile聚合成ManifestFile时partition信息会做聚合，如根据每个DataFile的partition value计算出ManifestFile的partition value的min-max信息，此外，ManifestFile还会记录增加/删除了多少数据。单个ManifestFile的大小默认设置为8M，这些ManifestFile会写成ManifestListFile，作为一个snapshot对应的文件。
2. Spark读Iceberg表
正如前面所介绍的，Iceberg存在很多Metadata文件，Spark在读Iceberg表时通过读这些Metadata文件可以实现高效的文件过滤。
首先根据partition summary进行文件过滤。如图，读取snapshot对应的ManifestFileList可以读到三个ManifestFile，然后根据where条件加上partition summary的min-max信息就可以过滤掉两个ManifestFile。这里的ManifestFile是Avro的文件，其中每条记录就是一个DataFile。根据每个DataFile的partition value和metrics信息还可以做进一步的过滤，最后只有三个文件需要进行真正的读取。
执行时Spark会将大的文件拆分成多个task，小的文件合并成一个task，每个task对应一到多个DataFile。因为Iceberg支持schema evolution，要读取的DataFile的schema和当前表的schema可能不匹配，因此需要做一个projection来保证返回的数据的schema和当前表的schema是相匹配的。
3. Iceberg文件过滤
前面提到，Iceberg是隐藏的分区，查询时不需要添加关于partition的筛选条件，那么它是怎么通过partition做到文件过滤的？
首先创建表，此时表的partition spec ID为0，接着写入数据，DataFile会记录写入数据的partition spec，此时是0。接着这里对表增加了基于天的partition field，此时表的partition spec ID变为1，然后再写入数据，这些数据的patition spec就是1。
在查询的时候，如图下方第一行是用户写的查询语句，但是Iceberg在真正的task plan时会额外加上partition filter，比如这里会对created_time做一个projection：对于spec为0的数据，使用month作为过滤条件；对于spec为1的数据，除了month的过滤条件还会加上day的过滤条件。
这些操作对用户都是透明的，即用户不需要关注具体的partition filed或者partition value来更改查询语句。
4. MOR-Position/Equality Delete
下面介绍一下Iceberg的upsert。Upsert有两种实现方式：copy on write和merge on read。
copy on write：实现方式类似Spark的overwrite
merge on read：Iceberg对merge on read的实现是写时生成DeleteFile，在读的时候将DeleteFile应用到DataFile上
DeleteFile有两种方式：Position Delete和Equality Delete。
（1）Position Delete
文件里记录的是哪个文件（file_path）的第几条记录（pos）需要被删除。
写Position Delete需要先读取DataFile，然后根据过滤条件判断哪些记录需要被删除，再写成Position DeleteFile。写入较慢，因为需要先进行task scan找到对应的文件，然后再写数据。
Spark的MOR目前只支持Position Delete。
（2）Equality Delete
文件里记录的是过滤条件，写入的速度快，读的速度慢，因为并不能准确的定位到文件，可能读了很多的DataFile，但是并不一定被删除，即apply的过程较慢。
Flink的CDC场景记录的就是这种方式。
5. Upsert-COW
这张图展示了在Iceberg中通过COW实现upsert的方式。
Upsert分为两部分：
找到需要更新的数据，将这些数据删掉
写入新数据
COW是很直观的过程，首先table scan会根据筛选条件找出所有需要更新的DataFile，接着在write的阶段会生成新的DataFile替换掉原来的DataFile，旧的DataFile被标记为删除。
6. Upsert-MOR
MOR第一阶段和COW类似，会根据筛选条件找到待更新的DataFile。
写入数据时不会删除这些旧的DataFile，只会生成新的DataFile，这种操作出现在update和upsert场景中。删除标记是通过DeleteFile来记录的，图中写出了两个Position DeleteFile记录DataFile中需要被删除的记录。
在读的时候会读Position DeleteFile应用到旧DataFile后的数据，加上新生成的DataFile，最后得到更新之后的结果。
--
03
Iceberg生产实践
1. 挑战1-宽表
这里选取了一个案例，就是腾讯的一个日志平台，希望用Iceberg对他们所有的日志文件做统一的管理，下面介绍在改造的过程中我们遇到的一些问题以及相应的解决方案。
前面已经介绍了Spark在write的时候会将所有的DataFile给collect到driver端，再进行commit操作。每个DataFile都需要记录很多列的metrics信息，如column_sizes、value_counts、min-max信息等，这些字段其实是map类型，所以理论上列的数量越多DataFile的体积就会越大。
我们遇到的情况就是日志平台的表会特别宽，有几千到几万个列，批量往Iceberg写数据时driver端直接就OOM了。
于是我们就做了一个优化，不是等所有的DataFile都collect到driver端才进行commit操作 。因为task有执行的先后，数据也不是同时到达，所以我们在driver端收集到一定数量的DataFile时就将它们写成一个ManifestFile，任务结束后driver端只会有一些ManifestFile，此时再把这些ManifestFile commit到Iceberg里面。
Iceberg也有两个table property用于设置是否记录列的metrics信息。
因为有些列不会作为用户读取数据时的筛选条件，不需要统计它们的metrics信息。对于这些不必要的列可以根据情况将这两个table property设置为none。
“write.metadata.metrics.column.col1”对col1列进行设置。
“write.metadata.metrics.default”对所有的列进行设置。
2. 挑战2-schema变动频繁
第二个挑战就是作为许多业务的下游，日志平台的数据来源的schema会经常变动，所以在写入Iceberg时会涉及到schema不匹配的情况。
Iceberg是支持schema evolution的，比较容易解决这种情况，但是对用户不太友好。从HDFS中读出来了一个DataFrame，需要先判断这个DataFrame的schema和Iceberg当前表的schema是否匹配，如果不匹配的话需要先update这个schema再进行写操作，这个过程就需要用户对Iceberg和Spark比较了解，而且这中间改还可能出现一些问题。
所以我们就做了“auto-merge-schema”这个feature：从Iceberg读数据到再写之前会自动判断DataFrame的schema和Iceberg的schema是否匹配，如果不匹配会先做merge，再写入数据。 这样用户就不需要关心写入数据的schema和表的schema是否匹配的问题，只需要把数据通过Spark的DataFrameWriter写进去就可以了。
目前Iceberg社区也已经支持写入时自动匹配schema，可以通过两种方式实现：
设置TableProperties:”write.spark.accept-any-schema”为true。如果对DataSourceV2 API比较熟悉的话就会知道， Spark默认在plan的时候会检查写入的DataFrame和表的schema是否匹配，不匹配就抛出异常。所以需要增加一个TableCapability（TableCapability.ACCEPT_ANY_SCHEMA），这样Spark就不会做这个检查，交由具体的DataSource来检查。
df.writeTo(tableName).option(“merge-schema”, “true”).XX。
3. 挑战3-Schema变动影响文件过滤
第三个就是schema evolution会影响文件过滤。支持schema evolution之后用户对schema变动比较多，可应用的场景也比较多。
比如像深度学习的场景，用户读取数据时往往并不需要读全部的列，有时会新增一些列，有时会删除一些列。我们在帮用户排查时发现了下面的问题：
如上图所示，首先蓝色的这三个DataFile是由schema0写入的，这个schema有两个列，分别是id和name。
这时候做了一个schema evolution，新增了一个列address，此时再写入了两个DataFile，然后做了一个filter查询，这个查询条件是id > 10并且address是start_with ‘some value’，之后进行table plan的时候会比较慢，因为读到了很多不相关的DataFile。
之所以会读到这些DataFile，前面介绍了Iceberg在做table plan的时候其实主要是根据min-max和partition value信息做文件过滤，但蓝色的三个DataFile中根本没有address信息，可以理解为都是none的，更没有min-max信息，所以Iceberg没法做判断，这些数据都会被读出来。
4. 基于Schema过滤文件
针对上述问题，我们基于schema做了一个文件过滤。 当写入ManifestFile和DataFile时新添加了SchemaID字段表示写入当前表时的schema，Iceberg table metadata中通过map结构存有SchemaID和schema的映射关系，比如图中SchemaID为0时表的schema是由id和name两个字段组成，SchemaID为1时新增了address字段，此外，还会存在当前表默认的SchemaID。将SchemaID信息加到ManifestFile和DataFile里有助于我们后续进行文件过滤。
举个例子， 这里有四个ManifestFile，前两个的Schema ID都是0，第三个是1，第四个是-1，-1表示文件的schema未知，即组成此ManifestFile的DataFile部分Schema ID为0，部分为1。在进行filter条件过滤时，筛选条件中包含了address字段，而Schema ID为0时不存在address字段，所以会过滤掉黄色的两个ManifestFile，还剩两个ManifestFile。第三个ManifestFile已经能保证其所有的DataFile的Schema ID都是1，因此就不需要做进一步的过滤了。对于Schema ID为 -1 的ManifestFile还需要做进一步的DataFile级别的过滤，这里就是把Schema ID为0的DataFile过滤掉，所以最终只需要读五个DataFile就可以了。
5. 其余优化项
刚才讲了一些具体的案例，现在介绍下我们围绕Spark读写Iceberg还做的一些别的优化。
①ZOrder优化文件布局
这个其实是比较常见的，大家比较熟悉的像是Iceberg，Hudi和Delta Lake都已经支持了，我们也用了一年多了，效果还不错。文件过滤很多都是基于min-max信息来做的，如果文件与文件之间尽可能不耦合的话就可以过滤掉更多的DataFile，减少需要读取的文件。ZOrder可以认为是sort的变种，在Spark中，单列的sort对文件的过滤是比较友好的，但如果sort by多列的话，会首先对column0做sort，然后column0相同的时候再做column1的sort，所以如果过滤的where条件是column1或者column2的时候文件过滤效果就不是很好。ZOrder过滤通过bit位的交互将多个column的sort by映射到一维空间，可以认为它们是线形的，这样一来filter条件中是filter column1还是column2或是column的任意组合，都在一个维度中，因此尽可能减少了不同DataFile之间min-max信息的重叠，可以过滤掉很多不需要读的文件，效果比较好。
②Parquet BloomFilter
Iceberg 目前还没有支持 Parquet 的 BloomFilter索引，我们内部已经实现了。通过table property来控制对哪些列开启Parquet BloomFilter索引，主要是对点查的加速。目前社区也已经有实现Patch了。
③Iceberg 索引
与Parquet BloomFilter不同，这里的Iceberg索引是独立于文件格式的索引，可以实现更多的索引加速。目前内部接近上线了。
④优化Parquet Vectorized Read Decimal
Iceberg的向量化读不是采用Spark读Parquet/ORC的那一套，而是基于Apache Arrow实现的。我们在跑TPCDS的时候发现Iceberg Parquet的向量化读的性能和Spark是有差距的，主要原因是在读取Decimal时差距比较大。这里我们对Decimal的向量化读做了优化，在1TB TPCDS数据集上，对store_sales表full table scan有近两倍的性能提升。
⑤多线程Plan Task，并发或者分布式的删除文件
早期版本的Iceberg plan task都是单线程的，当表的规模特别大，文件数量特别多的时候，性能就会急剧下降，还有像删除文件时也是，我们将它们都改成了并发或者分布式的实现。
⑥View的支持
这个是出于业务方的需要，因为Spark采用的是DataSourceV2 API，对view的支持不是很友好，从Spark3.3开始可能才只是有部分的API支持，实际目前还没有，我们内部已经自己实现了view的支持。
--
04
数据治理服务
后面介绍一下我们围绕Iceberg所做的数据治理服务。
1. 数据治理服务总览
这是一个数据治理服务的总览。数据通过MQ或者CDC之类的方式流入Iceberg表，中间这些都是当前基于Spark实现的异步的服务。Compaction Service是指自动合并小文件：主要监控表文件的数量，做一个异步的compaction。然后clean up metadata files，这个就是后面会介绍到的expire snapshot。Clustering Service主要指ZOrder这种，通过优化文件布局，可以更高效的进行文件过滤。此外，还有一些clean service。右边这里画的就是通过像BI之类的方式，向用户展示出表的一些情况。
2. Expire Snapshots
为什么要做expire snapshot，如果对Iceberg比较了解的话可能对此比较熟悉，那就是因为Iceberg会保留很多的snapshots，以上图为例。当前我们在snapshot n，此时查看HDFS文件大小是100GB，然后用户做了一个delete的操作将一个DataFile标记为删除了，snapshot变为n+1，但是用户这时候查HDFS会抱怨执行了删除操作后HDFS文件并没有减小，还是100G，所以就会产生一些误解。这里之所以DataFile被标记为delete但是没有被直接删除是因为在snapshot n的时候它还reference了DataFile。
接着他再做了一个更新操作后这个表变为120G，如果没有expire snapshot的话这个表真实存在于HDFS上的体积是线形增长的，会一直增加不会减少。Expire snapshot是Iceberg提供的表的管理工具，可以通过Spark SQL或者Spark具体的action来执行。比如这里expire掉历史的snapshot，只保留snapshot n+2，前面的snapshot都不要了，这时候没有snapshot reference被标记为删除的DataFile了，才将它真正的删除，这时候表的大小就变为80G了。
由于用户可能不太了解Iceberg的一些概念，不清楚该如何执行这一块，因此我们做了统一的数据治理服务，用户只需要把这个选项勾上就会自动的执行。
3. 合并小文件
合并小文件的原因是DataFile大小有可能不均匀。因为Iceberg是支持多种引擎写入的，比如Spark Batch，Spark Streaming或者Flink的方式，不同引擎写入的情况或者速度都不一样，所以DataFile本身有大小的区别，还有就是像CDC之类的场景会有一些DeleteFile，这些都会影响读的效率。
图中为合并小文件和clustering服务。首先就是BinPack，把大文件拆分成小文件，小文件合并成大文件，同时再apply上DeleteFile，是一个将原始DataFile通过combine写成差不多大小的DataFile，同时加一个merge on read的过程。
Sort对于列比较少、尤其是单列的情况，做文件过滤的效果比较好。还有前面介绍的ZOrder，通过clustering技术来做合并小文件同时优化文件布局。
4. 列生命周期管理
最后介绍一下列的生命周期管理。
Iceberg表有很多列，但是并非所有的列都有价值。比如上图中这个表有四个列，times是指被查询的次数，可以看到前三列被使用的频率很高，比如在深度学习或者机器学习的场景中，需要对模型进行训练，查询之后发现加了这个dog_color的值对训练模型的收敛等并不会有很大帮助，所以后来就不再查这一列了，也就是对业务真正有价值的只是前面的三列。而数据还是在源源不断地写入，第四列一直占用存储空间。于是我们就做了一个列的生命周期管理，就是在query的时候会把查询了哪些列报告给数据治理服务，数据治理服务会做一个统计，用户可以设置对应的列被查询多少次才认为它是有价值的。
比如用户设置的10次，而这个dog_color在一定的时间内被查询的次数不满足这个条件，就会触发一个Spark job，Spark job就会把这个column给删除，并把数据清理掉。目前我们主要采用的通过rewrite，把这些数据从物理上给删除。当然我们现在也在考虑通过像HBase这种column family来实现，就可以更高效的删除，但是这样就改了Parquet这种实现，可能会对不同引擎的兼容性有问题。
今天的分享就到这里，谢谢大家。
｜分享嘉宾｜
刘献杨
腾讯 高级软件工程师
刘献杨，目前供职于腾讯实时湖仓团队，主要参与基于Iceberg的湖仓研发和优化工作，曾就职于Intel大数据团队参与Spark优化和相关开发工作。活跃贡献于Iceberg/Spark/Ray等开源项目。
｜DataFun新媒体矩阵｜
｜关于DataFun｜
专注于大数据、人工智能技术应用的分享与交流。发起于2017年，在北京、上海、深圳、杭州等城市举办超过100+线下和100+线上沙龙、论坛及峰会，已邀请超过2000位专家和学者参与分享。其公众号 DataFunTalk 累计生产原创文章800+，百万+阅读，15万+精准粉丝。
