HUDI-0.11.0 BUCKET index on Flink 特性试用

1. 背景

在 0.10.1 版本下，使用默认的 index(FLINK_STATE)，在 upsert 模式下，几十亿级别的数据更新会消耗大量内存，并且检查点（checkpoint）时间过长。因此，切换到 0.11.0 的 BUCKET 索引。

当前环境：Flink 1.13.2 + Hudi 0.11.0（master 2022.04.11） + COW + HDFS。

关键配置项：

• index.type = BUCKET
• hoodie.bucket.index.num.buckets = 256

关键词

• Hudi
• COW
• Flink
• BUCKET
• FLINK_STATE

2. BUCKET 与 FLINK_STATE 的区别

FLINK_STATE

• 描述：Hudi 的 upsert 模式需要指定主键组，更新时按照主键进行更新。数据存储在 HDFS 文件上，因此需要维护主键与文件名的映射关系。Flink 的 state 用于存储这些映射关系。
• 特点：
  • 第一次加载 Hudi 表的历史数据时，需要设置 index.bootstrap.enabled = true 来加载历史数据到 state 中。
  • 支持跨分区更新。
• 缺点：
  • 占用大量内存。
  • 初始化加载历史数据较慢。

BUCKET

• 描述：基于文件的分桶机制。例如，设置主键为 id，桶个数为 256，则计算桶序号的方法为 (id.hashCode() & Integer.MAX_VALUE) % 256。
• 特点：
  • 桶的个数一旦设置，不能更改。
  • 文件个数固定，单个文件大小会随着数据量增加而增大。
  • 优点：不占用 Flink 的 Managed Memory。
  • 缺点：文件 IO 操作会增加 CPU 压力。
  • 不支持跨分区更新。
• 建议：
  • 预估数据量，设置合理的桶数量，避免小文件或写放大问题。
  • 可以通过离线导数据观察 HDFS 文件大小来预估桶数量。

总结

• FLINK_STATE：占用内存，初始化加载历史数据慢，支持跨分区。
• BUCKET：占用磁盘，不支持跨分区，节省内存。

3. 相关配置

Flink 实时流配置

'connector' = 'hudi',
'path' = 'hdfs://path/',
'index.type' = 'BUCKET',                 -- bucket索引
'hoodie.parquet.compression.codec'= 'snappy',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'upsert', 
'write.task.max.size' = '2048', 
'write.precombine' = 'true',
'write.precombine.field' = 'update_time',
'write.tasks' = '6',
'write.bucket_assign.tasks' = '6',
'hoodie.bucket.index.hash.field' = 'id',         -- 主键
'hoodie.bucket.index.num.buckets' = '256',  -- 桶个数
'hive_sync.enable'='true',
'hive_sync.table'='TABLE_NAME',
'hive_sync.db'='DB_NAME',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'thrift://HOST:9083',
'hive_sync.skip_ro_suffix' = 'true',
'write.insert.cluster' = 'true',
'write.ignore.failed' = 'true',
'clean.async.enabled' = 'true',
'clean.retain_commits' = '3', 
'hoodie.cleaner.commits.retained' = '3',
'hoodie.keep.min.commits' = '4', 
'hoodie.keep.max.commits' = '8'

Flink 离线导入数据配置

'connector' = 'hudi',
'path' = 'hdfs://PATH',
'hoodie.parquet.compression.codec'= 'snappy',
'index.type' = 'BUCKET',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'bulk_insert', 
'write.tasks' = '2', 
'hoodie.bucket.index.num.buckets' = '256', 
'hoodie.bucket.index.hash.field' = 'id'

离线导入完成后

• 观察 HDFS 文件，前八位为数字（例如 00000000-, 00000255-），即表示设置成功。
• 然后可直接接入实时数据。

注意

• 从 Hive 导数据到 Hudi 时，可以调整 Hive Source 的并行度：

  tableConfig.setInteger(HiveOptions.TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM_MAX, source_parallelism_max)
  

4. 性能小结

实时情况

• 基于当前数据量，单文件（80M）操作在 100ms 左右。例如：block read in memory in 171 ms. row count = 617384。
• 十几张表每次检查点（checkpoint）耗时约三四分钟，对于十几分钟的检查点间隔来说可以接受。

离线导数据情况

• 对于亿级别数据的离线导入，资源消耗不大，十几分钟即可完成。

注意事项

• 如果检查点设置过小，COW 表频繁操作 bucket 文件，会对集群 CPU 负载产生压力。