flink catalogs --- flink-1.12

Posted 2022-08-27 宝哥大数据

文章目录

• 一、Catalogs 类型
• • 1.1、GenericInMemoryCatalog
  • 1.2、JdbcCatalog
  • 1.3、HiveCatalog
  • 1.4、用户自定义 Catalog
• 三、Hive Catalog(重要)
• • 3.1、配置 hive catalog
  • 3.2、使用 Hive Catalog
• 四、Hive Read & Write
• 参考：

Catalog 提供了元数据信息，例如数据库、表、分区、视图以及数据库或其他外部系统中存储的函数和信息。

数据处理最关键的方面之一是管理元数据。 元数据可以是临时的，例如临时表、或者通过 TableEnvironment 注册的 UDF。 元数据也可以是持久化的，例如 Hive Metastore 中的元数据。Catalog 提供了一个统一的API，用于管理元数据，并使其可以从 Table API 和 SQL 查询语句中来访问。

一、Catalogs 类型

1.1、GenericInMemoryCatalog

GenericInMemoryCatalog 是基于内存实现的 Catalog，所有元数据只在 session 的生命周期内可用。

1.2、JdbcCatalog

JdbcCatalog 使得用户可以将 Flink 通过 JDBC 协议连接到关系数据库。PostgresCatalog 是当前实现的唯一一种 JDBC Catalog。 参考 JdbcCatalog 文档 获取关于配置 JDBC catalog 的详细信息。

-- 创建jdbc catalog
CREATE CATALOG mysql_catalog WITH(
    'type' = 'jdbc',
    'default-database' = 'flinktest',
    'username' = 'root',
    'password' = '123456',
    'base-url' = 'jdbc:mysql://chb1:3306'
);
 
-- 查看当前所有的catalog;
show catalogs;
 
-- 切换catalog
use catalog mysql_catalog;

1.3、HiveCatalog

HiveCatalog 有两个用途：作为原生 Flink 元数据的持久化存储，以及作为读写现有 Hive 元数据的接口。 Flink 的 Hive 文档 提供了有关设置 HiveCatalog 以及访问现有 Hive 元数据的详细信息。

警告 Hive Metastore 以小写形式存储所有元数据对象名称。而 GenericInMemoryCatalog 区分大小写。

1.4、用户自定义 Catalog

Catalog 是可扩展的，用户可以通过实现 Catalog 接口来开发自定义 Catalog。 想要在 SQL CLI 中使用自定义 Catalog，用户除了需要实现自定义的 Catalog 之外，还需要为这个 Catalog 实现对应的 CatalogFactory 接口。

CatalogFactory 定义了一组属性，用于 SQL CLI 启动时配置 Catalog。 这组属性集将传递给发现服务，在该服务中，服务会尝试将属性关联到 CatalogFactory 并初始化相应的 Catalog 实例。

三、Hive Catalog(重要)

3.1、配置 hive catalog

1、添加依赖
有两种添加 Hive 依赖项的方法。第一种是使用 Flink 提供的 Hive Jar包。您可以根据使用的 Metastore 的版本来选择对应的 Hive jar。第二个方式是分别添加每个所需的 jar 包。如果您使用的 Hive 版本尚未在此处列出，则第二种方法会更适合。

注意：建议您优先使用 Flink 提供的 Hive jar 包。仅在 Flink 提供的 Hive jar 不满足您的需求时，再考虑使用分开添加 jar 包的方式。

2、创建 hive catalog

catalogs:
   - name: myhive
     type: hive
     hive-conf-dir: /opt/hive-conf  # contains hive-site.xml

3.2、使用 Hive Catalog

1、创建一个 kafkaTable

Flink SQL> CREATE TABLE mykafka (name String, age Int) WITH (
   'connector.type' = 'kafka',
   'connector.version' = 'universal',
   'connector.topic' = 'test',
   'connector.properties.bootstrap.servers' = 'localhost:9092',
   'format.type' = 'csv',
   'update-mode' = 'append'
);
[INFO] Table has been created.

Flink SQL> DESCRIBE mykafka;
+------+--------+------+-----+--------+-----------+
| name |   type | null | key | extras | watermark |
+------+--------+------+-----+--------+-----------+
| name | STRING | true |     |        |           |
|  age |    INT | true |     |        |           |
+------+--------+------+-----+--------+-----------+

2、通过命令行向 kafka 写数据

localhost$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>tom,15
>john,21

3、在 Flink SQL client 查询结果

四、Hive Read & Write

https://blog.csdn.net/wuxintdrh/article/details/126296514

参考：

https://nightlies.apache.org/flink/flink-docs-release-1.12/zh/dev/table/catalogs.html
https://nightlies.apache.org/flink/flink-docs-release-1.12/zh/dev/table/connectors/hive/hive_catalog.html
https://nightlies.apache.org/flink/flink-docs-release-1.12/zh/dev/table/connectors/hive