Flink使用Hive Metastore管理元数据

浅浅的花香味﹌ 2023-05-22 10:24 29阅读 0赞

Flink sql连接kafka消费数据时，通常都要重复写create table，这就很让人烦恼。如果可以将这些DDL语句存储起来，下次直接使用就完美了。那Flink1.10版本提供了这个可能。

Hive作为大数据离线数仓的事实标准，有完善的元数据中心Hive Metastore。Flink1.10提供了HiveCatalog连接Hive Metastore，可以将kafka表元数据存储到Hive Metastore中。

*  1、集成 Hive 配置的依赖  
    因为使用SQL Client，需要将依赖的 jar 拷贝到 Flink 的 lib 目录中。

#hive版本1.2.1
    flink-connector-hive_2.11-1.10.0.jar
    hive-metastore-1.2.1.jar
    hive-exec-1.2.1.jar
    libfb303-0.9.2.jar

*  2、连接到Hive  
    首先需要到sql-client-defaults.yaml 中指定自己所需的 Catalog。

execution:
        planner: blink
        ...
        current-catalog: myhive  # set the HiveCatalog as the current catalog of the session
        current-database: mydatabase
        
    catalogs:
       - name: myhive
         type: hive
         hive-conf-dir: /opt/hive-conf
         hive-version: 1.2.1

其中：

1）name 是用户给每个 Catalog 实例指定的名字，Catalog 名字和 DB 名字构成了 FlinkSQL 中元数据的命名空间，因此需要保证每个 Catalog 的名字是唯一的。

2）type 表示 Catalog 的类型，对于 HiveCatalog 而言，type 应该指定为 hive。

3）hive-conf-dir 用于读取 Hive 的配置文件（hive-conf/hive-site.xml），可以将其设定为集群中 Hive 的配置文件目录。

3）hive-version 用于指定Hive 版本。

指定了 HiveCatalog 以后启动 sql-client，通过以下命令验证 HiveCatalog 已经正确加载。

show catalogs;
    use catalog myhive;
    show databases;
    show tables;
    create database ...;
    use database ...;
    create table kafkaTable...;
    select * from kafkaTable;