Apache Hudi 新版本新特性超强解读

本次分享主要是针对Apache Hudi 0.11.0新版本新特性进行深度解读，主要介绍4个方面的内容：

• 多级索引
• Spark SQL 新功能
• Flink 集成改进
• 其他功能和提升

01多级索引

首先和大家分享下多级索引，接下来我们将从三个方面介绍它。第一是为什么我们引入多级索引multi model index，第二多级索引的设计以及实践，最后将介绍如何利用多级索引，极大提升读写性能。

1. 为什么要引入多级索引（Multi-Modal Index）

在介绍多级索引之前，我们先看看索引是什么？索引是数据库系统中常用的查询加速技术。通过构建索引就可以利用生成的元数据metadata快速定位查询所需数据的位置，这样可以减少甚至避免从文件系统中扫描或者读取不必要的数据，减少IO的开销，大大提升查询效率。我们可以类比图书馆以及教科书中的索引，这些都是通过利用提前生成好的metadata来快速找到想要查找的信息。

其实在Apache Hudi的湖仓一体的架构中已经提供了特有的索引支持，这里我们可以看一个例子。其中我们用的一种索引可以快速定位到我们所需要更新或者删除的记录所在的文件组。如下图所示：

如果没有index的情况下，对于所有的更新以及删除的记录，我们需要和所有的文件进行merge，这样开销会非常大。如果使用了索引，读写开销会极大地降低，可以提高查询这个位置的效率。

在Hudi中我们默认开启了基于bloom filter的索引。这些bloom filter是存储在数据文件的footer中。作为一个单独的存在，它可以被用到索引的流程过程中。

那我们为什么还要在Hudi中引用多级索引？其实索引的主要目的就是刚刚提到的提升数据查询的速度，那么就需要存储metadata。而对于TB、EB级别表中的metadata，它们的量级会很大。通常的做法是把它们存储于单独的数据块block或者是数据文件中。而我们在实测的过程中，遇到了读写瓶颈。同时，在维护metadata的过程中，它需要实时的和数据表进行同步。这样的管理十分复杂，因为要保证transaction，如果我们要在Hudi中引用新的索引开发周期会很长。

2. 多级索引的设计与实现

为了解决上述的问题，我们在湖仓一体的存储架构中引入了多级索引，是首次在类似的架构中引入统一平台、多元化、高性能的索引。我们的目标是支持所有计算及查询引擎来提升读写性能，甚至未来如果出现新的引擎，也是可以进行兼容的。

接下来我们介绍一些在多级索引设计中所需要的需求。

• 第一是我们需要保证可拓展的元数据（scalable metadata）。我们希望元数据是Serverless的，是不需要任何的计算或者是内存中需要支持的，可以独立存在于数仓一体和数据湖中。同时希望它能独立于计算及查询引擎，它是可拓展的，能高性能地支持不同索引类型。
• 第二是我们希望多级索引中的元数据和数据表保持实时同步，保证每次的更新都是事务性的。
• 第三是保证查询的速度。保证对于多级索引的查询是低延迟的，主要的查询类型包括point, range, and prefix lookups等。

我们分别来看一下是如何实现的。

首先是可拓展的元数据，我们采用了和已有数据库类似的设计，那就是在内部构建一个元数据表meta table。对于Hudi table来说，我们利用的是Hudi的mor表来存储这些数据。mor表的优势是可以很快地进行数据的更新与删除。同时，Hudi表也是Serverless的，它不会依赖任何计算及内存资源。在这个表里我们针对不同的索引是建立独立的分区的。在这样的情况下，不同的index可以完成独立的管理以及自动化的管理。我们在使用mor表的另一个优点是可以支持任意的索引大小。从mb级别到gb级别再到tb级别。针对独立的分支，我们可以引入新的作用类型，就只需要建立新的分区。在构建可拓展的元数据的时候，需要索引的一个初始化。我们提高了两种方式的初始化。一种是同步，同步是在写入表的过程中，在最后commit之前会做一个index的步骤。而第二种方式是异步。异步创建索引是hudi首次引入的，保证了concurrent writer 不受影响。下面是异步创建索引的流程图：

• 第二点的设计原则是保证对metadata table的更新是事务性的，来保证metadata table结构里面的数据要和数据表实时同步。我们设计了一套叫multi table多表的transaction。同时在这个metadata table 里，有自我管理的表服务，包括compaction, cleaning。它们会保证定时操作，以保证这个metadata table 的读性能会很好。
• 第三点是对于metadata的快速查询。我们使用了HFile作为MDT的数据格式。原因是列格式Parquet或基于行的Avro不适合 pointed lookup；HFile格式中的index使得 pointed lookup非常高效，只需要读取相关数据块。我们针对HFile做了一个测试，得出在千万（10M）+条目中查询 N 条目，HFile 相比于 Parquet、Avro 有 10-100倍的提升。如下图：

3. 利用多级索引极大提升读写性能

接下来介绍多级索引所带来的主要的读写性能提升。

• 首先是File Listing

在云存储中我们发现大部分情况下，如果对于大型表的上千分区以及百万级的数据文件做listing，会造成读写瓶颈。这主要是因为云存储的设计所导致的。如果我们利用metadata table中的files来做分区。这个分区里提供了这个数据表里所有的file。相比于云软件系统有2-20倍的提升。如下图：

• 另一个比较重要的特性是Data Skipping

Data Skipping 技术是利用列统计数据来对所需要的这个数据文件做file pruning （文件裁剪），列统计数据常见的列统计数据包括取最大值、最小值、数量、大小等。Data Skipping 的作用就是通过这些统计数据来排除掉不需要读的文件，这样可以极大的提高查询速度。我们在这个multi model index 的metadata 中构建了column_stats分区，这个分区里的每条记录包含了这个Hudi表里所对应文件的列统计数据。每个Record key是由列名称、分区名称和文件名称组成。通过这种排列格式，可以快速定位所需的column stats。查询复杂度是基于查询列所需要的列的数量，通常这个数量是5到10个。对于大宽表来说，这样可以极大地提升这个效果。在实际测试中，云上Hudi大宽表的“定向”查询速度有10x-30x的提升。大幅减少了对无关数据文件的扫描和读取，提高了I/O效率。如下图：

• 我们还对Upsert 性能做了测试。我们在metadata table中引入了一个bloom_filter 分区，取代了footer 中的 bloom filter，在大表中可以大幅减少文件读取的时间。每条记录包含单个数据文件（partition + file name）的 bloom filter，支持 prefix lookup。据实测，在有 100k 文件的 Hudi 表中，相比于读取 footer，从 MDT 读取 bloom filter的速度有3倍的提升。t6.c
• 基于多级索引，未来还有很多工作可以做，目前一个工作是针对record level index 的开发以及Luncene index的开发。
  

02 Spark SQL新功能

接下来再讲Hudi在Spark SQL方面的改进。

1. 用户可以使用非主键字段更新或删除 Hudi 表中的记录

Spark SQL改进Delete Operation。

在t1时刻分别往mor表cow表中分别插入a,b,c三条数据。这样会在mor表中生成base file 文件和log file文件（下图中简化了示意图）。在cow表中只会生产base file 文件。t2时刻同时删除mor表和cow表中 b的数据。mor表操作是删除log file b的block是t2时刻的数据。而cow表中的操作是复制一份base file b保存到内存中，删除b数据之后会形成一个新版本的绿色的方框中的数据文件。如下图：

2. SQL支持时间旅行查询

我们为什么要实现Time travel？从api层面，如果我们要写一个查询，需要设置不同的df，构造不同的operation，然后来查询这个动作。但是引用time travel的这个语句以后，一是可以在spark sql中直接使用，二是sql语句更容易去解释这样的一个行为和动作。现在可以通过timestamp as of语法支持时间旅行查询，但仅限Spark 3.2+。语法如下：select * from hudi_tbl timestamp as of '20210728141108100' 

①SQL Travel-场景1：查询多版本数据

如下图，我们在10:10分提交了insert和update语句，我们想要查10:05分版本的数据，通过下面的sql是可以实现的。

select * from test_hudi timestamp as of 20220512100510000 （10:05）

select * from test_hudi timestamp as of 20220512101030000 （10:10）

②Travel-场景2： 数据还原修复

• 创建hudi表：create table test_hudi ...
• 插入数据：insert into test_hudi ... （每插入一条数据会产生一个版本）

• 查询数据：select * from test_hudi

• 误删数据：delete from test_hudi where id=2
• 查询数据：select * from test_hudi （删除id=2的数据后只剩下了两个版本）

• 还原数据：insert into test_hudi_table select * from test_hudi timestamp as of 20220511165343733（数据如果存在的话update，如果数据不存在就insert）
• 查询数据：select * from test_hudi （id=2的数据已经还原回来了）

注意：如果这个表用了 truncate清空的话，这种时间戳方式查询恢复就不行了。 

3. 添加CALL命令以支持在 Hudi 表上调用更多操作

Call Command产生的一个背景是spark sql除了ddl、dql和dml之外的操作，我们想解决这三种操作之外的一些新功能的操作。那么在引入Call Command之前是没办法操作的。我们对比了一下传统数据库里面的存储过程，类似地在spark这一块实现了一个command动作，然后对应实现了一个procedure的功能。

首先是在hudi一侧为call command生成了一个通用的语法，不依赖于spark的版本，可以对所有spark版本适用。然后生成了一个HoodieProcedure的类，使用CallProcedureHoodieCommand类调用动作。

Call Command命令在设计时主要有四个方面的功能。一是支持归档、提交、回滚和创建还原点的快照动作。二是可以进行原数据管理。三是对运维表进行数据导入导出、Boostrap、修复表、升级/降级等操作。四是优化表动作，如Compaction、Clustering、Clean等。

Call Command的参数有三类。一是可以使用不定式的参数（键值对）作为它的入参参数。二是可以按照参数的位置进行入参。三是混合参数。以下是传参的具体语法：

• 接下来是CALL Command功能方面的一些介绍。

①CALL Command-快照管理的相应的命令

• 查询快照 ：call show_commits_metadata(table => 'test_hudi_table', limit => 10)
• 回滚快照：

call rollback_to_instant('test_hudi_table',20220511224632307')

• 创建保存点 ：create_savepoints
• 恢复保存点 ：rollback_savepoints
• 删除保存点 ：delete_savepoints

②CALL Command-Clustering

可以设置Clustering的类型：

set hoodie.layout.optimize.strategy=linear /z-order/hilbert

常用命令：

call run_clustering(table => 'test_hudi_table',order => 'ts') 

call show_clustering(table=>'test_hudi_table')

通过这些clustering的动作，在查询的时候性能能达到10-20倍的提升。

③CALL Command-Compact（小文件合并）（目前只支持mor表）

Data file和Delta log file 合并会重新生成一个新的文件。

命令：call run_compaction(table=>'test_hudi_table',op=> instant

03 Flink集成改进

最后，介绍Flink集成改进方面的内容。主要有以下几点：

1. 在 0.11.0 中，同时支持 Flink 1.13.x 和 1.14.x。

2. 支持复杂的数据类型。

3. 基于 DFS 的 Flink HoodieCatalog。

绿色是目前已经实现的操作，红色是待实现的操作。可以通过ApI实现或者是Flink Sql实现。

FlinkHoodieCatalog在三个方面有比较好的作用：

• 第一个方面是对元数据的管理，通过FlinkHoodieCatalog这个框架可以更好地管理HoodieCatalog；
• 第二个方面是可以基于这个框架可以去和外面的其他的平台去对接，更方便对接使用；
• 第三个方面是元数据方面，可以在数据血缘方面构造一些功能。

4. Bucket Index 

为什么要集成改进Bucket Index？这个是字节的同学贡献的一个功能。他们在他们的生产场景里面，在34tb的数据量在5000亿的记录写入的情况下，Bloom Filter Index 通过Record key去找File ID 的这样的一个动作的性能会下降得很快。为了解决Bloom Filter Index的假阳性，他们引入了Bucket Index。

通过key的哈希值定位到File Group，提升了实时导入的性能。如下图所示：

从Flink输入了5条数据，然后通过一定的哈希策略将混合的Bucket Index进行关联，通过拿到FileGroupId写入文件。

• 利用Bucket分布做优化

Bucket分布优化主要有：Bucket Pruning、Bucket Aggregate、Bucket Join等。如下图所示：

• Bucket限制。目前 Bucket Index 的桶数量 ，需要根据预估的数据量提前在建表时进行确定，且建表后不可更改。
• Bucket使用

参数：hoodie.index.type 值：BUCKET

参数：hoodie.bucket.index.num.buckets 值：48（256MB）

建议单个桶的大小控制在3GB左右。

• Bucket后续改进。Hashmap的扩容过程，将分桶数按倍数做到轻量级动态扩容。

04其他功能和提升

最后，我们来讲其他功能和提升。

1. Spark DataSource 查询优化

我们在使用mor表做快照查询的时候，log文件会被读取，然后和base文件进行合并。在之前的版本中，当你做快照查询的时候，整条log文件记录会被读出来。这个版本我们做了优化，使用了内置的标准Payload来读取。例如：OverwriteWithLatestAvroPayload。我们会针对这个做了优化，只把必要的列读出来，这样就会极大的减少内存和压缩解码带来的CPU的消耗。其实是对于非常宽的上千列的表来说，效果会非常明细。

2. Schema 演进 

在这个版本中，我们针对Spark 3.1、Spark 3.2版本增加了schema功能的演进。如果启用 set hoodie.schema.on.read.enable=true以后，我们可以对表列和对表进行一系列的操作。列的变更（增加、删除、重命名、修改位置、修改属性），表的变更（重命名、修改属性） 等。

3. 保存点和恢复 

保存点和恢复可以用call command做这些操作。新版本引进了mor表，用Hudi CLI设定保存点和执行恢复或者call command来手动设置保存点。保存点之后的数据将会被删除。

4. Pulsar 写提交回调 

• Hudi写入commit支持Pulsar下游作业* 
• 主要配置 HoodieWriteCommitPulsarCallback 

hoodie.write.commit.callback.pulsar.topic

hoodie.write.commit.callback.pulsar.broker.service.url

• 可扩展的接口HoodieWriteCommitCallback 

5. 多元化的Catalog同步

• Google BigQuery 支持查询COW表* 
• DataHub 支持同步Schema 
• 通过AWS SDK原生同步Glue Data Catalog