自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

Flink SQL通過Hudi HMS Catalog讀寫Hudi并同步Hive表

作者：董可倫
數(shù)據(jù)庫 其他數(shù)據(jù)庫
本文介紹了Flink SQL如何通過Hudi HMS Catalog讀寫Hudi并同步Hive表，并且講述了Hudi HMS Catalog的好處，我認為這是目前比較完美的一種方式，強烈推薦大家使用。

?前言

上篇文章??Flink SQL操作Hudi并同步Hive使用總結(jié)??總結(jié)了如何使用Flink SQL讀寫Hudi并同步Hive，介紹了創(chuàng)建表的各種方式，但是每一種方式都不太完美。本文介紹一種比較完美的方式，通過Hudi HMS Catalog讀寫Hudi并同步Hive表，這里的Hudi HMS Catalog實際上就是通過上篇文章最后提到的HoodieHiveCatalog?實現(xiàn)的，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創(chuàng)建表一樣，直接將表建立在Hive中，并且表結(jié)構(gòu)與Hive SQL和Spark SQL兼容，也就是Flink Hudi HMS Catalog中創(chuàng)建的表，可以同時使用Flink SQL、Hive SQL、Spark SQL查詢，也可以同時使用Flink SQL、Spark SQL寫Hudi。不像上篇文章中介紹的方式，F(xiàn)link SQL寫Hudi的表不能被Hive/Spark使用，只能通過同步表的方式。另外在Flink Hudi HMS Catalog中和Spark SQL一樣默認開啟同步Hive,也就是對于MOR表默認會同步創(chuàng)建對應(yīng)的_ro表和_rt表，至于COW表因為同步的表名和創(chuàng)建的表名一樣，所以讀寫是同一張表?？傊蚐park SQL創(chuàng)建表、讀寫一致。

版本

Flink 1.14.3Hudi  master 0.13.0-SNAPSHOT。

創(chuàng)建Flink Hudi HMS Catalog

先看一下如何創(chuàng)建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中創(chuàng)建一個數(shù)據(jù)庫test_flink
create database if not exists hudi_catalog.test_flink;
## 切換到數(shù)據(jù)庫test_flink
use hudi_catalog.test_flink;

支持的配置項:

catalog.path
default-database
hive.conf.dir
# 可選項hms、dfs
mode
property-version
# 0.12.1版本應(yīng)該還不支持，需要自己拉取master最新代碼，PR支持：https://github.com/apache/hudi/pull/6923
# 是否為外部表，默認false，也就是默認內(nèi)部表
# 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，也就是默認內(nèi)部表，但是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相關(guān)的配置，也就是默認會同步，但仍然需要配置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中已經(jīng)配置了hive.conf.dir，這倆其實可以共用的。

這時在對應(yīng)的Hive數(shù)據(jù)庫中就已經(jīng)建好表了,并且表結(jié)構(gòu)同時兼容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幾條數(shù)據(jù)，看一下會不會觸發(fā)一下Hive同步。

insert into test_hudi_flink_mor values (1,'hudi',10,100,'2022-10-31'),(2,'hudi',10,100,'2022-10-31');

果然默認同步，表結(jié)構(gòu)和之前的方式是一樣的。同步的表默認是外部表，可以通過配置項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中的表結(jié)構(gòu),路徑為庫路徑+表名：hdfs://cluster1/warehouse/tablespace/managed/hive/test_flink/test_hudi_flink_cow，這更符合平時的使用習(xí)慣，畢竟少了一個配置項，且路徑統(tǒng)一好管理，不容易出錯。

+----------------------------------------------------+
|                   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');

因為名字一樣，所以同步的結(jié)果看不到變化。

一致性驗證

通過Spark SQL分別往每個表寫幾條數(shù)據(jù)，再用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');

經(jīng)過驗證，一致性沒有問題。遺憾的是，F(xiàn)link SQL查詢結(jié)果依舊不包含元數(shù)據(jù)字段，不清楚為啥要這樣設(shè)計~

異常解決

異常信息

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環(huán)境下的jetty版本不一致，導(dǎo)致有沖突，相關(guān)PR：https://github.com/apache/hudi/pull/6844?,這個PR升級了jetty的版本。解決思路，使hadoop環(huán)境下的jetty版本和Hudi包中的版本一致。一個方法是使Flink任務(wù)不依賴Hadoop環(huán)境下的jetty相關(guān)的jar，這里是由于配置了HADOOP_CLASSPATH，經(jīng)過嘗試一時無法解決。另外一個是升級Hadoop環(huán)境下的jetty版本，但是我嘗試了一下，由于Hadoop環(huán)境組件依賴的jar包比較多，單純升級jetty版本的話，會引起其他問題，無奈只能先將Hudi中jetty回退到原先的版本，最簡單的方式是直接reset到這個PR之前的位置。(先跑通Hudi HMS Catalog，后面有時間再解決依賴沖突問題)。

圖片

總結(jié)

本文介紹了Flink SQL如何通過Hudi HMS Catalog讀寫Hudi并同步Hive表，并且講述了Hudi HMS Catalog的好處，我認為這是目前比較完美的一種方式，強烈推薦大家使用。

責(zé)任編輯：武曉燕 來源： 倫少的博客
SQL
相關(guān)推薦

2022-11-01 07:43:30

2022-12-08 07:17:49

2022-10-28 07:10:51

HudiJavaHive

2021-08-31 10:07:16

Flink Hud數(shù)據(jù)湖阿里云

2023-02-26 00:12:10

Hadoop數(shù)據(jù)湖存儲

2024-04-26 07:36:42

Hudi 1.0數(shù)據(jù)湖倉數(shù)據(jù)查詢

2022-06-09 14:19:46

順豐數(shù)據(jù)集成Flink

2022-10-24 00:26:51

大數(shù)據(jù)Hadoop存儲層

2021-09-13 13:46:29

Apache HudiB 站數(shù)據(jù)湖

2022-10-17 07:51:31

Hudi異常HDFS

2013-09-16 16:20:55

自動備份Dropbox

2022-12-23 16:52:22

Lakehouse數(shù)據(jù)湖

2022-07-20 11:47:18

數(shù)據(jù)

2020-03-26 10:05:18

大數(shù)據(jù)IT互聯(lián)網(wǎng)

2023-07-19 16:22:00

Hudi機器學(xué)習(xí)

2022-10-17 10:48:50

Hudi大數(shù)據(jù)Hadoop

2021-09-13 14:19:03

HudiLakehouse阿里云

2022-06-08 13:25:51

數(shù)據(jù)

2021-01-06 10:36:55

MySQL數(shù)據(jù)庫Hive

2023-09-05 07:22:17

Hudi數(shù)據(jù)存儲
點贊
收藏

51CTO技術(shù)棧公眾號