11、Flink Table API & SQL - Catalogs（使用概览和样例）

今天药忘吃喽~ 2024-04-17 06:01 64阅读 0赞

Catalogs提供了元数据，例如数据库、表、分区、视图、函数和访问存储在数据库或其他外部系统中的数据所需的信息。

数据处理最重要的方面之一是管理元数据。它可以是临时表之类的临时元数据，也可以是针对表环境注册的UDF函数。或者永久的元数据，比如Hive中的元数据。Catalogs提供了一个统一的API来管理元数据，并可以通过表API和SQL查询。

## Catalog Types ##

**GenericInMemoryCatalog**

Flink sessions总是有一个名为default\_catalog的内置GenericInMemoryCatalog，它有一个名为default\_database的内置默认数据库。所有临时元数据，如使用TableEnvironment\#registerTable定义的表，都注册到这个catalog中。

**HiveCatalog**

HiveCatalog有两个用途：作为纯Flink元数据的持久存储，并作为读取和写入现有Hive元数据的接口。Flink的Hive文档提供了关于设置Catalog和与现有Hive安装接口的完整细节（详见[Flink Table API & SQL - Hive之hive概览][Flink Table API _ SQL - Hive_hive]）。

**警告：Hive元数据以小写存储所有元对象名称。`GenericInMemoryCatalog区分大小写。`**

**自定义Catalog**

用户可以通过实现Catalog接口开发自定义Catalog。要在SQL CLI中使用自定义Catalog，用户应该通过实现CatalogFactory接口开发Catalog及其对应的Catalog Factory。

catalog factory定义了一组属性，用于SQL CLI启动时初始化Catalog配置。这组属性将传递给discovery service，在该服务中，将尝试将属性与CatalogFactory匹配，并启动相应的Catalog实例。

## Catalog API ##

**Registering a Catalog（注册一个Catalog）**

用户可以在现有的Flink会话中注册额外的Catalog。

tableEnv.registerCatalog(new CustomCatalog("myCatalog"));

使用YAML定义的所有目录必须提供指定目录类型的类型属性。以下类型支持开箱即用。

<table> 
 <tbody> 
  <tr> 
   <th>Catalog</th> 
   <th>Type Value</th> 
  </tr> 
  <tr> 
   <td>GenericInMemory</td> 
   <td>generic_in_memory</td> 
  </tr> 
  <tr> 
   <td>Hive</td> 
   <td>hive</td> 
  </tr> 
 </tbody> 
</table>

catalogs:
       - name: myCatalog
         type: custom_catalog
         hive-conf-dir: ...

**Changing the Current Catalog And Database（更改当前Catalog和数据库）**

Flink将一直搜索当前目录和数据库中的表、视图和UDF。

tableEnv.useCatalog("myCatalog");
    tableEnv.useDatabase("myDb");

Flink SQL> USE CATALOG myCatalog;
    Flink SQL> USE myDB;

通过在catalog.database.object表单中提供完全限定名，可以访问来自非当前编目的编目的元数据。

tableEnv.scan("not_the_current_catalog", "not_the_current_db", "my_table");

Flink SQL> SELECT * FROM not_the_current_catalog.not_the_current_db.my_table;

**列出可用的Catalog、数据库、表**

tableEnv.listCatalogs();
    
    tableEnv.listCatalogs();
    
    tableEnv.listTables();

Flink SQL> show catalogs;
    
    Flink SQL> show databases;
    
    Flink SQL> show tables;

[Flink Table API _ SQL - Hive_hive]: https://blog.csdn.net/zpf_940810653842/article/details/100065614