Hudi On Flink

一、背景

传统数仓分为离线和实时两个部分

1. 离线部分属于业务驱动，固定的计算逻辑，通过定时调度，最后产出报表；
2. 实时部分属于需求驱动，需要灵活开发。

传统架构整体还是以离线为主，实时为辅，实时指标的开发是粗放的，没有schema的规范，没有元数据的管理，也没有打通实时和离线数据之间的联系，但两者实际上解决的都是相同的业务问题，最大的区别就在于时效性。

二、实时数仓建设

1. 首先统一数仓标准，元数据和开发流程。
2. 引入Hudi 加速数仓宽表，基于Flink SQL建设实时数仓，
3. 加强平台治理，进行数仓平台化建设，实现数据统一接入，统一开发和统一的元数据管理。

实时数仓方案对比

1. Lambda 架构
   Lambda 架构是在原有离线数仓的基础上，将对实时性要求比较高的部分剥离出来，增加了一个实时速度层。Lambda 架构的缺点是需要维护实时和离线两套架构和两套开发逻辑，维护成本比较高，另外两套架构带来的资源消耗也是比较大的。
2. Kappa 架构
   Kappa 架构移除了原有的离线部分，使用纯流式引擎开发。Kappa 架构的最大问题是，流数据重放处理时的吞吐能力达不到批处理的级别，导致重放时产生一定的延时。

引入Hudi 加速宽表

决定改造原有 Lambda 架构，通过加速它的离线部分来建设数仓宽表。此时，就需要一个工具来实时快速的更新和删除 Hive 表，支持 ACID 特性，支持历史数据的重放。市面上的三款开源组件：Delta Lake、Iceberg、Hudi，最后选择 Hudi 来加速宽表。

Hudi 关键特性：

1. 可回溯历史数据
2. 支持大规模数据根据主键更新删除数据
3. 支持数据增量消费
4. 支持HDFS小文件压缩

首先，在ODS层进入Hudi实现实时数据接入，将ODS层T+1的全量数据抽取改成T+0的实时接入，从数据源头实现Hive 表的加速
另外，使用Flink消费Kafka中接入的数据，进行清洗聚合，通过Hudi 增量更新DWD层的Hive 宽表，将宽表从离线加速成准实时。

最终架构

引入Hudi后，基于Lambda架构，定制化的实时数仓最终架构如下图所示。
实时速度层通过CDC接入数据到Kafka，采用Flink SQL处理Kafka中的数据，并将ODS层Kafka数据清洗计算后通过Hudi准实时更新DWD层的宽表，以加速宽表的产出。离线层采用Hive存储及处理。最后由 ADS 层提供统一的数据存储与服务。

image.png

三、Hudi On Flink 的原理

这里介绍下 Hudi On Flink 的原理。Hudi 原先与 Spark 强绑定，它的写操作本质上是批处理的过程。为了解耦 Spark 并且统一 API ，Hudi On Flink 采用的是在 Checkpoint 期间攒批的机制，在 Checkpoint 触发时将这一批数据Upsert 到 Hive，根据 Upsert 结果统一提交或回滚。

Hudi On Flink 的实现流可以分解为几个步骤：

1. 首先使用 Flink 消费 Kafka 中的 Binlog 类型数据，将其转化为 Hudi Record。
2. Hudi Record 进入 InstantTime Generator，该 Operator 并不对数据做任何处理，只负责转发数据。它的作用是每次 Checkpoint 时在 Hudi 的 Timeline 上生成全局唯一且递增的 Instant，并下发。
3. 随后，数据进入 Partitioner ，根据分区路径以及主键进行二级分区。分区后数据进入 File Indexer ，根据主键找到在 HDFS 上需要更新的对应文件，将这个对应关系按文件 id 进行分桶，并下发到下游的 WriteProcessOperator 。
4. WriteProcessOperator 在 Checkpoint 期间会积攒一批数据，当 Checkpoint 触发时，通过 Hudi 的 Client 将这批数据 Upsert 到 HDFS 中，并且将 Upsert 的结果下发到下游的 CommitSink 。
5. CommitSink 会收集上游所有算子的 upsert 结果，如果成功的个数和上游算子的并行度相等时，就认为本次 commit 成功，并将 Instant 的状态设置为 success ，否则就认为本次 commit 失败并进行回滚。

四、Hudi On Flink 优化

• 二级分区
  对于增量写入的场景，大部分的数据都写入当天的分区，可能会导致数据倾斜。因此，我们使用分区路径和主键 id 实现二级分区，避免攒批过程中单个分区数据过多，解决数据倾斜问题。

• 文件索引
  Hudi 写入过程的瓶颈在于如何快速找到记录要写入的文件并更新。为此 Hudi 提供了一套索引机制，该机制会将一个记录的键 + 分区路径的组合映射到一个文件 ID. 这个映射关系一旦记录被写入文件组就不会再改变。Hudi 当前提供了 HBase、Bloom Filter 和内存索引 3 种索引机制。然而经过生产实践，HBase 索引需要依赖外部的组件，内存索引可能存在 OOM 的问题，Bloom Filter 存在一定的误算率。

在 Hudi 写入的 parquet 文件中存在一个隐藏的列，通过读取这个列可以拿到文件中所有数据的主键，因此可以通过文件索引获取到数据需要写入的文件路径，并保存到 Flink 算子的 state 中，也避免了外部依赖和 OOM 的问题。

• 索引写入分离
  原先 Hudi 的 Upsert 过程，写入和索引的过程是在一个算子中的，算子的并行度只由分区路径来决定。将索引和写入的过程进行分离，这样可以提高 Upsert 算子的并行度，提高写入的吞吐量。

• 故障恢复
  整个流程的状态保存到 Flink State 中，设计了一套基于 State 的故障恢复机制，可以保证端到端的 exactly-once 语义。