麒麟v10 上部署 TiDB v5.1.2 生产环境优化实践
1776
2023-04-22
Flink SQL通过Hudi HMS Catalog读写Hudi并同步Hive表
前言
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_flinkcreate database if not exists hudi_catalog.test_flink;## 切换到数据库test_flinkuse hudi_catalog.test_flink;
支持的配置项:
可以看到和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默认值为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读写。
同步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,这更符合平时的使用习惯,毕竟少了一个配置项,且路径统一好管理,不容易出错。
同步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查询结果依旧不包含元数据字段,不清楚为啥要这样设计~
异常解决
异常信息
总结
本文介绍了Flink SQL如何通过Hudi HMS Catalog读写Hudi并同步Hive表,并且讲述了Hudi HMS Catalog的好处,我认为这是目前比较完美的一种方式,强烈推荐大家使用。
版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。