FlinkSQL通过HudiHMSCatalog读写Hudi并同步Hive表

​前言

上篇文章​​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。

创新互联建站专注于企业全网整合营销推广、网站重做改版、新民网站定制设计、自适应品牌网站建设、H5开发商城网站建设、集团公司官网建设、成都外贸网站制作、高端网站制作、响应式网页设计等建站业务,价格优惠性价比高,为新民等各大城市提供网站开发制作服务。

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 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.(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的好处,我认为这是目前比较完美的一种方式,强烈推荐大家使用。

本文名称:FlinkSQL通过HudiHMSCatalog读写Hudi并同步Hive表
分享链接:http://www.shufengxianlan.com/qtweb/news40/151840.html

网站建设、网络推广公司-创新互联,是专注品牌与效果的网站制作,网络营销seo公司;服务项目有等

广告

声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联