1. 树叶云首页
  2. 技术资讯

Flink SQL通过Hudi HMS Catalog读写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。

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表

建表

CREATETABLE 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 createtable test_hudi_flink_mor;
## 可以自己验证一下table.external是否生效
+----------------------------------------------------+
|                   createtab_stmt                   |
+----------------------------------------------------+
|CREATETABLE `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切换
CREATETABLE 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查询。

insertinto test_hudi_flink_mor values(3,'hudi',10,100,'2022-10-31');
insertinto test_hudi_flink_mor_ro values(4,'hudi',10,100,'2022-10-31');
insertinto test_hudi_flink_mor_rt values(5,'hudi',10,100,'2022-10-31');
insertinto 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的好处，我认为这是目前比较完美的一种方式，强烈推荐大家使用。

文章来源网络，作者：运维，如若转载，请注明出处：https://shuyeidc.com/wp/236949.html<

赞 (0)
运维的头像运维
00
上一篇2025-04-22 06:15
下一篇 2025-04-22 06:17

相关推荐

  • 如何制作个人主题技术资讯

    个人主题怎么制作？

    制作个人主题是一个将个人风格、兴趣或专业领域转化为视觉化或结构化内容的过程，无论是用于个人博客、作品集、社交媒体账号还是品牌形象，核心都是围绕“个人特色”展开，以下从定位、内容规划、视觉设计、技术实现四个维度,详细拆解制作个人主题的完整流程，明确主题定位：找到个人特色的核心主题定位是所有工作的起点，需要先回答……

    运维的头像运维
    2025-11-20
    0
  • 如何管理社群营销技术资讯

    社群营销管理关键是什么？

    社群营销的核心在于通过建立有温度、有价值、有归属感的社群，实现用户留存、转化和品牌传播，其管理需贯穿“目标定位-内容运营-用户互动-数据驱动-风险控制”全流程,以下从五个维度展开详细说明：明确社群定位与目标社群管理的首要任务是精准定位，需明确社群的核心价值（如行业交流、产品使用指导、兴趣分享等）、目标用户画像……

    运维的头像运维
    2025-11-20
    0
  • 香港公司如何网站备案技术资讯

    香港公司网站备案需要什么材料？

    香港公司进行网站备案是一个涉及多部门协调、流程相对严谨的过程，尤其需兼顾中国内地与香港两地的监管要求，由于香港公司注册地与中国内地不同，其网站若主要服务内地用户或使用内地服务器，需根据服务器位置、网站内容性质等，选择对应的备案路径（如工信部ICP备案或公安备案），以下从备案主体资格、流程步骤、材料准备、注意事项……

    运维的头像运维
    2025-11-20
    0
  • 如何企业上云推广技术资讯

    如何企业上云推广

    企业上云已成为数字化转型的核心战略,但推广过程中需结合行业特性、企业痛点与市场需求，构建系统性、多维度的推广体系，以下从市场定位、策略设计、执行落地及效果优化四个维度，详细拆解企业上云推广的实践路径，精准定位：明确目标企业与核心价值企业上云并非“一刀切”的方案，需先锁定目标客户群体，提炼差异化价值主张，客户分层……

    运维的头像运维
    2025-11-20
    0
  • PS如何设计搜索框技术资讯

    PS设计搜索框的实用技巧有哪些？

    在PS中设计一个美观且功能性的搜索框需要结合创意构思、视觉设计和用户体验考量，以下从设计思路、制作步骤、细节优化及交互预览等方面详细说明，帮助打造符合需求的搜索框，设计前的规划明确使用场景：根据网站或APP的整体风格确定搜索框的调性，例如极简风适合细线条和纯色，科技感适合渐变和发光效果，电商类则可能需要突出搜索……

    运维的头像运维
    2025-11-20
    0

发表回复

您的邮箱地址不会被公开。必填项已用 * 标注