?前言
上篇文章??Flink SQL操作Hudi并同步Hive使用總結??總結了如何使用Flink SQL讀寫Hudi并同步Hive，介紹了創建表的各種方式，但是每一種方式都不太完美。本文介紹一種比較完美的方式，通過Hudi HMS Catalog讀寫Hudi并同步Hive表，這里的Hudi HMS Catalog實際上就是通過上篇文章最后提到的HoodieHiveCatalog?實現的，PR：https://github.com/apache/hudi/pull/6082,2022年7月18 merge，也就是從Hudi0.12.0版本開始支持(我確認了一下0.11.1版本沒有)，如果大家要使用的話，必須升級到0.12.0+，本文使用Hudi master 0.13.0-SNAPSHOT。
Flink Hudi HMS Catalog的好處
既然推薦這種方式，那么先說一下它的好處吧。好處是它可以像Spark SQL創建表一樣，直接將表建立在Hive中，并且表結構與Hive SQL和Spark SQL兼容，也就是Flink Hudi HMS Catalog中創建的表，可以同時使用Flink SQL、Hive SQL、Spark SQL查詢，也可以同時使用Flink SQL、Spark SQL寫Hudi。不像上篇文章中介紹的方式，Flink SQL寫Hudi的表不能被Hive/Spark使用，只能通過同步表的方式。另外在Flink Hudi HMS Catalog中和Spark SQL一樣默認開啟同步Hive,也就是對于MOR表默認會同步創建對應的_ro表和_rt表，至于COW表因為同步的表名和創建的表名一樣，所以讀寫是同一張表。總之和Spark SQL創建表、讀寫一致。
版本
Flink 1.14.3Hudi master 0.13.0-SNAPSHOT。
創建Flink Hudi HMS Catalog
先看一下如何創建Flink Hudi HMS Catalog。
CREATE CATALOG hudi_catalog WITH (
'type' = 'hudi',
'mode' = 'hms',
'default-database' = 'default',
'hive.conf.dir' = '/usr/hdp/3.1.0.0-78/hive/conf',
'table.external' = 'true'
);
## 其實就是在Hive中創建一個數據庫test_flink
create database if not exists hudi_catalog.test_flink;
## 切換到數據庫test_flink
use hudi_catalog.test_flink;
支持的配置項:
catalog.path
default-database
hive.conf.dir
# 可選項hms、dfs
mode
property-version
# 0.12.1版本應該還不支持，需要自己拉取master最新代碼，PR支持：https://github.com/apache/hudi/pull/6923
# 是否為外部表，默認false，也就是默認內部表
# 0.12.0和0.12.1沒有這個配置項，只能是外部表
table.external
可以看到和hive catalog的配置項差不多，只是type為hudi，這里mode必須是hms，默認值是dfs，至于為啥是hms，請看下面的源碼分析還有一點需要注意的是hive catalog中的配置項為hive-conf-dir，但是hudi的為hive.conf.dir，看著差不多，其實不一樣。table.external：是否為外部表，默認false，也就是默認內部表，但是0.12.0和0.12.1沒有這個配置項，只能是外部表，這正是我使用Hudi master 0.13.0-SNAPSHOT的原因如果覺得這個配置不是必須的，大家可以直接用0.12.1即可。
為啥mode為hms
public Catalog createCatalog(Context context) {
final FactoryUtil.CatalogFactoryHelper helper =
FactoryUtil.createCatalogFactoryHelper(this, context);
helper.validate();
String mode = helper.getOptions().get(CatalogOptions.MODE);
switch (mode.toLowerCase(Locale.ROOT)) {
case "hms":
return new HoodieHiveCatalog(
context.getName(),
(Configuration) helper.getOptions());
case "dfs":
return new HoodieCatalog(
context.getName(),
(Configuration) helper.getOptions());
default:
throw new HoodieCatalogException(String.format("Invalid catalog mode: %s, supported modes: [hms, dfs].", mode));
}
}
public static final ConfigOption<String> MODE = ConfigOptions
.key("mode")
.stringType()
.defaultValue("dfs"); 可以看到mode默認值為dfs,只有mode為hms時，才會使用HoodieHiveCatalog。
MOR表
建表
CREATE TABLE test_hudi_flink_mor (
id int PRIMARY KEY NOT ENFORCED,
name VARCHAR(10),
price int,
ts int,
dt VARCHAR(10)
)
PARTITIONED BY (dt)
WITH (
'connector' = 'hudi',
'path' = '/tmp/hudi/test_hudi_flink_mor',
'table.type' = 'MERGE_ON_READ',
'hoodie.datasource.write.keygenerator.class' = 'org.apache.hudi.keygen.ComplexAvroKeyGenerator',
'hoodie.datasource.write.recordkey.field' = 'id',
'hoodie.datasource.write.hive_style_partitioning' = 'true',
'hive_sync.conf.dir'='/usr/hdp/3.1.0.0-78/hive/conf'
);
使用catalog時path可以不用指定，不指定的話，路徑就是Hive庫路徑+表名，可以看后面的Cow表。
這里需要注意的是，雖然不用配置同步Hive相關的配置，也就是默認會同步，但仍然需要配置hive_sync.conf.dir，否則依舊會報和上篇文章中一樣的異常：WARN hive.metastore [] - set_ugi() not successful, Likely cause: new client talking to old server. Continuing without it.org.apache.thrift.transport.TTransportException: null其實這里我認為是不合理的，因為catalog中已經配置了hive.conf.dir，這倆其實可以共用的。
這時在對應的Hive數據庫中就已經建好表了,并且表結構同時兼容Hive、Spark和Flink,也就是既可以用Hive SQL查詢，也可以用Spark SQL和Flink SQL讀寫。
show create table test_hudi_flink_mor;
## 可以自己驗證一下table.external是否生效
+----------------------------------------------------+
| createtab_stmt |
+----------------------------------------------------+
| CREATE TABLE `test_hudi_flink_mor`( |
| `_hoodie_commit_time` string, |
| `_hoodie_commit_seqno` string, |
| `_hoodie_record_key` string, |
| `_hoodie_partition_path` string, |
| `_hoodie_file_name` string, |
| `id` int, |
| `name` string, |
| `price` int, |
| `ts` int) |
| PARTITIONED BY ( |
| `dt` string) |
| ROW FORMAT SERDE |
| 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' |
| WITH SERDEPROPERTIES ( |
| 'hoodie.query.as.ro.table'='false', |
| 'path'='/tmp/hudi/test_hudi_flink_mor', |
| 'primaryKey'='id', |
| 'type'='mor') |
| STORED AS INPUTFORMAT |
| 'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat' |
| OUTPUTFORMAT |
| 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' |
| LOCATION |
| 'hdfs://cluster1/tmp/hudi/test_hudi_flink_mor' |
| TBLPROPERTIES ( |
| 'connector'='hudi', |
| 'hive_sync.conf.dir'='/usr/hdp/3.1.0.0-78/hive/conf', |
| 'hoodie.datasource.write.hive_style_partitioning'='true', |
| 'hoodie.datasource.write.keygenerator.class'='org.apache.hudi.keygen.ComplexAvroKeyGenerator', |
| 'hoodie.datasource.write.recordkey.field'='id', |
| 'path'='/tmp/hudi/test_hudi_flink_mor', |
| 'spark.sql.create.version'='spark2.4.4', |
| 'spark.sql.sources.provider'='hudi', |
| 'spark.sql.sources.schema.numPartCols'='1', |
| 'spark.sql.sources.schema.numParts'='1', |
| 'spark.sql.sources.schema.part.0'='{"type":"struct","fields":[{"name":"id","type":"integer","nullable":false,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"price","type":"integer","nullable":true,"metadata":{}},{"name":"ts","type":"integer","nullable":true,"metadata":{}},{"name":"dt","type":"string","nullable":true,"metadata":{}}]}', |
| 'spark.sql.sources.schema.partCol.0'='dt', |
| 'table.type'='MERGE_ON_READ', |
| 'transient_lastDdlTime'='1667373370') |
+----------------------------------------------------+
同步Hive
Insert幾條數據，看一下會不會觸發一下Hive同步。
insert into test_hudi_flink_mor values (1,'hudi',10,100,'2022-10-31'),(2,'hudi',10,100,'2022-10-31');
果然默認同步，表結構和之前的方式是一樣的。同步的表默認是外部表，可以通過配置項hoodie.datasource.hive_sync.create_managed_table配置是否為外部表。
COW 表
建表
## 建表時可以直接catalog.database.table,不用use切換
CREATE TABLE hudi_catalog.test_flink.test_hudi_flink_cow (
id int PRIMARY KEY NOT ENFORCED,
name VARCHAR(10),
price int,
ts int,
dt VARCHAR(10)
)
PARTITIONED BY (dt)
WITH (
'connector' = 'hudi',
'hoodie.datasource.write.keygenerator.class' = 'org.apache.hudi.keygen.ComplexAvroKeyGenerator',
'hoodie.datasource.write.recordkey.field' = 'id',
'hoodie.datasource.write.hive_style_partitioning' = 'true',
'hive_sync.conf.dir'='/usr/hdp/3.1.0.0-78/hive/conf'
);
這里沒有指定path，看一下Hive中的表結構,路徑為庫路徑+表名：hdfs://cluster1/warehouse/tablespace/managed/hive/test_flink/test_hudi_flink_cow，這更符合平時的使用習慣，畢竟少了一個配置項，且路徑統一好管理，不容易出錯。
+----------------------------------------------------+
| createtab_stmt |
+----------------------------------------------------+
| CREATE EXTERNAL TABLE `test_hudi_flink_cow`( |
| `_hoodie_commit_time` string, |
| `_hoodie_commit_seqno` string, |
| `_hoodie_record_key` string, |
| `_hoodie_partition_path` string, |
| `_hoodie_file_name` string, |
| `id` int, |
| `name` string, |
| `price` int, |
| `ts` int) |
| PARTITIONED BY ( |
| `dt` string) |
| ROW FORMAT SERDE |
| 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' |
| WITH SERDEPROPERTIES ( |
| 'hoodie.query.as.ro.table'='true', |
| 'path'='hdfs://cluster1/warehouse/tablespace/managed/hive/test_flink/test_hudi_flink_cow', |
| 'primaryKey'='id') |
| STORED AS INPUTFORMAT |
| 'org.apache.hudi.hadoop.HoodieParquetInputFormat' |
| OUTPUTFORMAT |
| 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' |
| LOCATION |
| 'hdfs://cluster1/warehouse/tablespace/managed/hive/test_flink/test_hudi_flink_cow' |
| TBLPROPERTIES ( |
| 'connector'='hudi', |
| 'hive_sync.conf.dir'='/usr/hdp/3.1.0.0-78/hive/conf', |
| 'hoodie.datasource.write.hive_style_partitioning'='true', |
| 'hoodie.datasource.write.keygenerator.class'='org.apache.hudi.keygen.ComplexAvroKeyGenerator', |
| 'hoodie.datasource.write.recordkey.field'='id', |
| 'path'='hdfs://cluster1/warehouse/tablespace/managed/hive/test_flink/test_hudi_flink_cow', |
| 'spark.sql.create.version'='spark2.4.4', |
| 'spark.sql.sources.provider'='hudi', |
| 'spark.sql.sources.schema.numPartCols'='1', |
| 'spark.sql.sources.schema.numParts'='1', |
| 'spark.sql.sources.schema.part.0'='{"type":"struct","fields":[{"name":"id","type":"integer","nullable":false,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"price","type":"integer","nullable":true,"metadata":{}},{"name":"ts","type":"integer","nullable":true,"metadata":{}},{"name":"dt","type":"string","nullable":true,"metadata":{}}]}', |
| 'spark.sql.sources.schema.partCol.0'='dt', |
| 'transient_lastDdlTime'='1667375710') |
+----------------------------------------------------+
同步Hive
insert into test_hudi_flink_cow values (1,'hudi',10,100,'2022-10-31'),(2,'hudi',10,100,'2022-10-31');
因為名字一樣，所以同步的結果看不到變化。
一致性驗證
通過Spark SQL分別往每個表寫幾條數據，再用Spark、Hive、Flink查詢。
insert into test_hudi_flink_mor values (3,'hudi',10,100,'2022-10-31');
insert into test_hudi_flink_mor_ro values (4,'hudi',10,100,'2022-10-31');
insert into test_hudi_flink_mor_rt values (5,'hudi',10,100,'2022-10-31');
insert into test_hudi_flink_cow values (3,'hudi',10,100,'2022-10-31');
經過驗證，一致性沒有問題。遺憾的是，Flink SQL查詢結果依舊不包含元數據字段，不清楚為啥要這樣設計~
異常解決
異常信息
Caused by: java.lang.NoSuchMethodError: org.apache.hudi.org.apache.jetty.util.compression.DeflaterPool.ensurePool(Lorg/apache/hudi/org/apache/jetty/util/component/Container;)Lorg/apache/hudi/org/apache/jetty/util/compression/DeflaterPool;
at org.apache.hudi.org.apache.jetty.websocket.server.WebSocketServerFactory.<init>(WebSocketServerFactory.java:184) ~[hudi-flink1.14-bundle-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT]
異常原因，Hudi包中的jetty版本和hadoop環境下的jetty版本不一致，導致有沖突，相關PR：https://github.com/apache/hudi/pull/6844?,這個PR升級了jetty的版本。解決思路，使hadoop環境下的jetty版本和Hudi包中的版本一致。一個方法是使Flink任務不依賴Hadoop環境下的jetty相關的jar，這里是由于配置了HADOOP_CLASSPATH，經過嘗試一時無法解決。另外一個是升級Hadoop環境下的jetty版本，但是我嘗試了一下，由于Hadoop環境組件依賴的jar包比較多，單純升級jetty版本的話，會引起其他問題，無奈只能先將Hudi中jetty回退到原先的版本，最簡單的方式是直接reset到這個PR之前的位置。(先跑通Hudi HMS Catalog，后面有時間再解決依賴沖突問題)。
總結
本文介紹了Flink SQL如何通過Hudi HMS Catalog讀寫Hudi并同步Hive表，并且講述了Hudi HMS Catalog的好處，我認為這是目前比較完美的一種方式，強烈推薦大家使用。
