• flink catalogs --- flink-1.12


    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
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    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
    
    • 1
    • 2
    • 3
    • 4

    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 |     |        |           |
    +------+--------+------+-----+--------+-----------+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

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

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

    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

  • 相关阅读:
    [ERROR Swap]: running with swap on is not supported. Please disable swap
    这个c++程序到底错在哪里了?
    PyCharm连接MySQL数据库竟然如此简单
    【Linux】25. 生产者消费者模型
    Flask 学习-93.cookie 有效期设置
    算法: 编辑距离使得单词相同 72. Edit Distance
    一文深入浅出理解国产开源木兰许可系列协议
    Qt常用快捷键
    Vue-router路由
    Qt 工程添加windows库文件
  • 原文地址:https://blog.csdn.net/wuxintdrh/article/details/126257030