Apache Hudi 建表需要考虑哪些参数Spark-- 上篇

整体流程

• 选择表类型、主键、预合并字段
• 设置 clean 和 archive 参数，对于 mor 表设置 compact 参数
• 选择合适的索引
• 设置合适的 存储/写入参数
• 确认建表注意事项

建表

基本参数

选择合适的表类型

hudi 目前支持两种表类型，分别是 cow 表和 mor 表，详情见 https://hudi.apache.org/docs/table_types

举个栗子简单介绍两者个实现区别：

• 假设有一张表，已经有 100 个列式存储 parquet 数据文件，每个文件 100M
• 这时候需要更新写入一条数据，hudi 会根据这条数据的主键 id 寻找需要更新哪些文件，假设需要更新文件 A
  • 对于 cow 类型，就需要重新生成整个 100M 的文件 A（带上新的数据替代旧数据），会有写放大问题
  • 对于 mor 类型，只需要生成或者追加到行式文件 log，算是预写入，等写入到一定的大小或者次数才会和 cow 表一样重写文件 A。解决了写放大问题，在一定程度升有读放大问题。

选择合适的主键 primaryKey

主键在 hudi 中是一等公民，需要提供主键以支持去重等功能，并且确保同一张表或者分区下不会有主键重复的数据

支持多个字段，逗号分隔

选择合适的 preCombineField

hudi 根据主键去重，需要判断是否将旧的数据更新掉，会通过 primaryKey 进行判断，比较大就更新，一般是 ts 时间戳。如果不设置，默认每次都会用最新的数据更新替换

是否创建外表

建表时指定 location 才会以 hive 中外表的方式进行处理，如果 sql 只有 external 没有 location，依然按内表处理

基本 sql case

create table hudi_cow_pt_tbl (
  id bigint,
  name string,
  ts bigint,
  dt string,
  hh string
) using hudi
options (
  type = 'cow',
  primaryKey = 'id',
  preCombineField = 'ts'
 )
partitioned by (dt, hh)
location '/tmp/hudi/hudi_cow_pt_tbl';

其它常用参数

注意：这里只会列出需要用户配置或者常用的，其它的可以访问 https://hudi.apache.org/docs/configurations

文件组织管理（原理介绍）

这里先介绍下 hudi 的文件组织管理（cow）

从离线任务简单分析，如图下图，假设任务 5 分钟运行一次，每次会生成一个 commit

• 10:00，包含提交了 file_1 和 file_2 蓝色的部分，生成两个文件
• 10:05，任务提交生成了绿色三个文件，对于 file_1 和 file_2 生成的是新的版本文件，包含 10:00 新的数据或者是相同主键更新的数据
• 10:10，粉色部分类似的原理

这时候，file_1 有三个版本，2 也有三个，3有 1 个，4有一个

然后，根据 timeline 和 多个版本的数据文件，基于此我们就可以实现附加的功能

• 时间旅行，可以通过 sql 查询 绿色版本的数据，也可以查询蓝色版本的数据，甚至可以做 diff
• 增量查询，目前 sql 不支持，可以查询 10:05 ~ 10:10 的增量数据
• 回滚：可以回滚到任何一个 active timeline 上的 commit 时间点，再进行数据增量修补

上面提到 active timeline，就有非 active timeline，叫做 archive timeline，hudi 会根据其参数进行归档。所有以上的附加功能基于的都是 active timeline，所以要合理控制 active timeline 的大小

可能有疑问，我需要历史全量数据岂不是要保持所有 timeline 为 active。hudi 会保证每个 file 至少一个版本，所以不用保持所有 timeline，毕竟数据存储成本会比较高。

了解了上面的原理，那么就需要通过一定的参数配置解决某些问题：

• 比如保留多少数据版本防止文件过多？
• 保留多少 active timeline 防止元数据过大？

Clean Config

clean 配置用来清理不需要的数据文件版本，减少数据冗余，默认每次任务完成都会触发判断是否需要清理

Archive Config

archive 用来将 active timeline 进行归档，防止 active timline 过大影响 hudi 元数据访问的性能

Compaction Config

对于 mor 表，还有一项特殊的属性，就是 compact config，配置 mor 表的 log 和 base file 的合并策略，mor 组织结构如下：

分为两个文件结构：

• Columnor Base file：存储历史已经合并的文件
• Row based Append Log：新到的数据直接追加到 log 文件中，不对 base file 重写，读取时和 base file 进行合并
  

存储/写入属性

Storage/Write Config

其它

索引属性

索引使用场景有两个，一个是快速定位 recordKey 所在的文件，一个是加速数据查询，减少 scan 的数据文件，这里涉及的是快速定位 recordKey

注意：对于分区表，非 global 索引只能确保分区能唯一

进阶属性（待补充，可以先参考官方文档使用）

• Multy writer Multi Writer Guarantees 解决多个 writer 并发写入的冲突问题
• Metadata Configs Metadata Table hudi 元数据表，对 bloom 数据读取和查询方式进行加锁
• Full Schema evolution Schema Evolution 支持灵活的 schema变更
• Locks Configuration 结合 Multy writer 使用
• Memory Configurations
• Bootstrap Configs

建表注意点

1. 首次写入数据没有历史数据评估行大小，hoodie.copyonwrite.record.size.estimate 的配置可能过大
   1. 先写入 100M 大小的数据，再全量写入
   2. 或者通过 parquet 文件分析行大小进行设定
2. 合理设置 active time 的数量大小
   1. 主要用户增量读取构建下游表
   2. 也用于数据回滚、清理、合并
3. 对于分区表，非 global 索引只能确保分区能唯一

案例 case

COW 表

• 保留最近 48 次的提交，假设你的任务每小时跑一次，那么等于保存近两天的 commit，下游可以消费最近 48 小时的增量数据
• 同时也要修改 archive 归档参数，最小的 commits 不能低于要保留的 commit count
• 假设我已经分析 parquet 知道每行数据在 parquet 中的存储大小为 10 b（考虑到 hoodie 自己的元数据字段，大头是 record key，和表自定的主键长度有关）
• 分区名称保持和 hive 一致
• 外表

create table if not exists h3(
  id bigint,
  name string,
  price double,
  ts bigint,
  dt string
) using hudi
options (
  primaryKey = 'id',
  type = 'cow',
  preCombineField = 'ts',
  hoodie.cleaner.policy = 'KEEP_LATEST_COMMITS',
  hoodie.cleaner.commits.retained = '48',
  hoodie.keep.min.commits = '50',
  hoodie.keep.max.commits = '60',
  hoodie.copyonwrite.record.size.estimate = '10',
  hoodie.datasource.write.hive_style_partitioning = 'true'
)
partitioned by (dt)
location '/user/prod_datalake_test/datalake_test/hive/h3';

MOR 表

对于 mor 表来说，compact 时机影响 ro 表的数据新鲜度，对于 ro 查询只能查询 base file，不会查询 log file （Query types），并且可以提升查询性能

• 保留最近 48 次的提交，假设你的任务每小时跑一次，那么等于保存近两天的 commit，下游可以消费最近 48 小时的增量数据
• 同时也要修改 archive 归档参数，最小的 commits 不能低于要保留的 commit count
• 假设我已经分析 parquet 知道每行数据在 parquet 中的存储大小为 10 b（考虑到 hoodie 自己的元数据字段，大头是 record key，和表自定的主键长度有关）
• 分区名称保持和 hive 一致
• 外表
• 设置 12 小时合并一次，并且在任务写入完成后触发

注意：12 小时 合并一次，代表 第 12 次倍数的周期任务运行时间必定会比其它周期任务执行时间长，如果有一定的忧虑，可以进行异步合并计划生成，异步合并计划执行

create table if not exists h3(
  id bigint,
  name string,
  price double,
  ts bigint,
  dt string
) using hudi
options (
  primaryKey = 'id',
  type = 'mor',
  preCombineField = 'ts',
  hoodie.cleaner.policy = 'KEEP_LATEST_COMMITS',
  hoodie.cleaner.commits.retained = '48',
  hoodie.keep.min.commits = '50',
  hoodie.keep.max.commits = '60',
  hoodie.copyonwrite.record.size.estimate = '10',
  hoodie.datasource.write.hive_style_partitioning = 'true',
  hoodie.compact.inline.trigger.strategy = 'TIME_ELAPSED',
  hoodie.compact.inline.max.delta.seconds = '43200'
)
partitioned by (dt)
location '/user/prod_datalake_test/datalake_test/hive/h3';

这篇好文章是转载于：学新通技术网