Hudi Catalog
快速体验 PALO & Hudi
使用限制
- Hudi 表支持的查询类型如下：
| 表类型 | 支持的查询类型 |
|---|---|
| Copy On Write | Snapshot Query, Time Travel, Icremental Read |
| Merge On Read | Snapshot Queries, Read Optimized Queries, Time Travel, Icremental Read |
- 目前支持 Hive Metastore 和兼容 Hive Metastore 类型 (例如AWS Glue/Alibaba DLF) 的 Catalog。
创建 Catalog
和 Hive Catalog 基本一致，这里仅给出简单示例。
1CREATE CATALOG hudi PROPERTIES (
2 'type'='hms',
3 'hive.metastore.uris' = 'thrift://172.21.0.1:7004',
4 'hadoop.username' = 'hive',
5 'dfs.nameservices'='your-nameservice',
6 'dfs.ha.namenodes.your-nameservice'='nn1,nn2',
7 'dfs.namenode.rpc-address.your-nameservice.nn1'='172.21.0.2:4007',
8 'dfs.namenode.rpc-address.your-nameservice.nn2'='172.21.0.3:4007',
9 'dfs.client.failover.proxy.provider.your-nameservice'='org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider'
10);可选配置参数：
| 参数名 | 说明 | 默认值 |
|---|---|---|
| use_hive_sync_partition | 使用 hms 已同步的分区数据 | false |
列类型映射
和 Hive Catalog 一致，可参阅 Hive Catalog 中 列类型映射 一节。
Skip Merge
Spark 在创建 hudi mor 表的时候，会创建 _ro 后缀的 read optimize 表，PALO 读取 read optimize 表会跳过 log 文件的合并。PALO 判定一个表是否为 read optimize 表并不是通过 _ro 后缀，而是通过 hive inputformat，用户可以通过 SHOW CREATE TABLE 命令观察 cow/mor/read optimize 表的 inputformat 是否相同。
此外 PALO 支持在 catalog properties 添加 hoodie 相关的配置，配置项兼容 Spark Datasource Configs。所以用户可以在 catalog properties 中添加 hoodie.datasource.merge.type=skip_merge 跳过合并 log 文件。
查询优化
PALO 使用 parquet native reader 读取 COW 表的数据文件，使用 Java SDK(通过 JNI 调用 hudi-bundle) 读取 MOR 表的数据文件。在 upsert 场景下，MOR 依然会有数据文件没有被更新，这部分文件可以通过 parquet native reader 读取，用户可以通过 explain 命令查看 hudi scan 的执行计划，hudiNativeReadSplits 表示有多少 split 文件通过 parquet native reader 读取。
1|0:VHUDI_SCAN_NODE |
2| table: minbatch_mor_rt |
3| predicates: `o_orderkey` = 100030752 |
4| inputSplitNum=810, totalFileSize=5645053056, scanRanges=810 |
5| partition=80/80 |
6| numNodes=6 |
7| hudiNativeReadSplits=717/810 |用户可以通过 profile 查看 Java SDK 的性能，例如：
1- HudiJniScanner: 0ns
2 - FillBlockTime: 31.29ms
3 - GetRecordReaderTime: 1m5s
4 - JavaScanTime: 35s991ms
5 - OpenScannerTime: 1m6sOpenScannerTime: 创建并初始化 JNI Reader 的时间JavaScanTime: Java SDK 读取数据的时间FillBlockTime: Java 数据拷贝为 C++ 数据的时间GetRecordReaderTime: 调用 Java SDK 并创建 Hudi Record Reader 的时间
Time Travel
每一次对 Hudi 表的写操作都会产生一个新的快照，Time Travel 支持读取 Hudi 表指定的 Snapshot。默认情况下，查询请求只会读取最新版本的快照。
可以使用 FOR TIME AS OF 语句，根据快照的时间 (时间格式和 Hudi 官网保持一致) 读取历史版本的数据。示例如下：
1SELECT * FROM hudi_tbl FOR TIME AS OF "2022-10-07 17:20:37";
2SELECT * FROM hudi_tbl FOR TIME AS OF "20221007172037";
3SELECT * FROM hudi_tbl FOR TIME AS OF "2022-10-07";Hudi 表不支持 FOR VERSION AS OF 语句，使用该语法查询 Hudi 表将抛错。
Incremental Read
Incremental Read 可以查询在 startTime 和 endTime 之间变化的数据，返回的结果集是数据在 endTime 的最终状态。
PALO 提供了 @incr 语法支持 Incremental Read:
1SELECT * from hudi_table@incr('beginTime'='xxx', ['endTime'='xxx'], ['hoodie.read.timeline.holes.resolution.policy'='FAIL'], ...);beginTime 是必须的，时间格式和 hudi 官网 hudi_table_changes保持一致，支持 "earliest"。endTime 选填，默认最新 commitTime。兼容 Spark Read Options。
支持 Incremental Read 需要开启新优化器，新优化器默认打开。通过 desc 查看执行计划，可以发现 PALO 将 @incr 转化为 predicates 下推给 VHUDI_SCAN_NODE:
1| 0:VHUDI_SCAN_NODE(113) |
2| table: lineitem_mor |
3| predicates: (_hoodie_commit_time[#0] >= '20240311151019723'), (_hoodie_commit_time[#0] <= '20240311151606605') |
4| inputSplitNum=1, totalFileSize=13099711, scanRanges=1 |