Hudi（19）：Hudi集成Flink之索引和Catalog

0. 相关文章链接

** Hudi文章汇总 **

1. Bucket索引（从 0.11 开始支持）

    默认的 flink 流式写入使用 state 存储索引信息：primary key 到 fileId 的映射关系。当数据量比较大的时候，state的存储开销可能成为瓶颈，bucket 索引通过固定的 hash 策略，将相同 key 的数据分配到同一个 fileGroup 中，避免了索引的存储和查询开销。

1.1. WITH参数

名称

Required

默认值

说明

index.type

false

FLINK_STATE

设置 BUCKET 开启 Bucket 索引功能

hoodie.bucket.index.hash.field

false

主键

可以设置成主键的子集

hoodie.bucket.index.num.buckets

false

4

默认每个 partition 的 bucket 数，当前设置后则不可再变更。

1.2. 和 state 索引的对比

（1）bucket index 没有 state 的存储计算开销，性能较好
（2）bucket index 无法扩 buckets，state index 则可以依据文件的大小动态扩容
（3）bucket index 不支持跨 partition 的变更(如果输入是 cdc 流则没有这个限制)，state index 没有限制

2. Hudi Catalog（从 0.12.0 开始支持）

2.1. 概述

通过 catalog 可以管理 flink 创建的表，避免重复建表操作，另外 hms 模式的 catalog 支持自动补全 hive 同步参数。

DFS 模式 Catalog SQL样例：

CREATE CATALOG hoodie_catalog
  WITH (
    'type'='hudi',
    'catalog.path' = '${catalog 的默认路径}',
    'mode'='dfs' 
  );

Hms 模式 Catalog SQL 样例：

CREATE CATALOG hoodie_catalog
  WITH (
    'type'='hudi',
    'catalog.path' = '${catalog 的默认路径}',
    'hive.conf.dir' = '${hive-site.xml 所在的目录}',
    'mode'='hms' -- 支持 'dfs' 模式通过文件系统管理表属性
  );

2.2. WITH 参数

名称

Required

默认值

说明

catalog.path

true

--

默认的 catalog 根路径，用作表路径的自动推导，默认的表路径：${catalog.path}/${db_name}/${table_name}

default-database

false

default

默认的 database 名

hive.conf.dir

false

--

hive-site.xml 所在的目录，只在 hms 模式下生效

mode

false

dfs

支持 hms模式通过 hive 管理元数据

table.external

false

false

是否创建外部表，只在 hms 模式下生效

2.3. 使用dfs方式

步骤一：创建sql-client初始化sql文件

vim /opt/module/flink-1.13.6/conf/sql-client-init.sql

CREATE CATALOG hoodie_catalog
  WITH (
    'type'='hudi',
    'catalog.path' = '/tmp/hudi_catalog',
    'mode'='dfs' 
  );

USE CATALOG hoodie_catalog;

步骤二：指定sql-client启动时加载sql文件（重点：-i conf/sql-client-init.sql）

hadoop fs -mkdir /tmp/hudi_catalog

bin/sql-client.sh embedded -i conf/sql-client-init.sql -s yarn-session

步骤三：建库建表插入

create database test;
use test;

create table t2(
  uuid varchar(20),
  name varchar(10),
  age int,
  ts timestamp(3),
  `partition` varchar(20),
primary key (uuid) not enforced
)
with (
  'connector' = 'hudi',
  'path' = '/tmp/hudi_catalog/default/t2',
  'table.type' = 'MERGE_ON_READ'
);

insert into t2 values('1','zs',18,TIMESTAMP '1970-01-01 00:00:01','a');

步骤四：退出sql-client，重新进入，表信息还在

use test;
show tables;
select * from t2;

注：**其他Hudi相关文章链接由此进 -> Hudi文章汇总 **