카탈로그

카탈로그 (Catalogs)

Catalogs는 데이터베이스, 테이블, 파티션, 뷰, 함수와 같은 메타데이터와 데이터베이스나 다른 외부 시스템에 저장된 데이터에 접근하는 데 필요한 정보를 제공합니다. 데이터 처리에서 가장 중요한 측면 중 하나는 메타데이터를 관리하는 것입니다. Catalogs는 메타데이터를 관리하고 Table API와 SQL 쿼리에서 접근할 수 있게 하는 통합 API를 제공합니다.

출처: 문서

본문

Catalogs는 데이터베이스, 테이블, 파티션, 뷰, 함수 같은 메타데이터와 데이터베이스나 다른 외부 시스템에 저장된 데이터에 접근하는 데 필요한 정보를 제공합니다.

데이터 처리의 가장 중요한 측면 중 하나는 메타데이터를 관리하는 것입니다. 이는 임시 테이블이나 테이블 환경에 등록된 UDF 같은 일시적인 메타데이터일 수 있습니다. 또는 Hive Metastore의 영구 메타데이터일 수 있습니다. Catalogs는 메타데이터를 관리하고 Table API와 SQL 쿼리에서 접근할 수 있게 하는 통합 API를 제공합니다.

Catalog는 사용자가 데이터 시스템의 기존 메타데이터를 참조하고 이를 Flink의 해당 메타데이터에 자동으로 매핑할 수 있게 합니다. 예를 들어 Flink는 JDBC 테이블을 Flink 테이블에 자동으로 매핑할 수 있으며, 사용자는 Flink에서 DDL을 수동으로 다시 작성할 필요가 없습니다. Catalog는 사용자의 기존 시스템으로 Flink를 시작하는 데 필요한 단계를 크게 단순화하고 사용자 경험을 크게 향상시킵니다.

Catalog 유형 (Catalog Types)

GenericInMemoryCatalog

GenericInMemoryCatalog는 카탈로그의 인메모리 구현입니다. 모든 객체는 세션의 수명 동안에만 사용할 수 있습니다.

JdbcCatalog

JdbcCatalog는 사용자가 JDBC 프로토콜을 통해 Flink를 관계형 데이터베이스에 연결할 수 있게 합니다. Postgres Catalog와 MySQL Catalog가 현재 JDBC Catalog의 유일한 두 구현입니다. 카탈로그 설정에 대한 자세한 내용은 JdbcCatalog 문서를 참조하세요.

HiveCatalog

HiveCatalog는 두 가지 목적을 수행합니다. 순수 Flink 메타데이터를 위한 영구 저장소이자, 기존 Hive 메타데이터를 읽고 쓰기 위한 인터페이스입니다. Flink의 Hive 문서는 카탈로그 설정과 기존 Hive 설치와의 인터페이스에 대한 전체 세부 정보를 제공합니다.

Hive Metastore는 모든 메타 객체 이름을 소문자로 저장합니다. 이는 대소문자를 구분하는 GenericInMemoryCatalog와 다릅니다.

사용자 정의 Catalog (User-Defined Catalog)

Catalogs는 플러그형이며 사용자는 Catalog 인터페이스를 구현해 사용자 정의 카탈로그를 개발할 수 있습니다.

Flink SQL에서 사용자 정의 카탈로그를 사용하려면 사용자는 CatalogFactory 인터페이스를 구현해 해당 카탈로그 팩토리를 구현해야 합니다. 팩토리는 Java의 Service Provider Interfaces(SPI)를 사용해 발견됩니다. 이 인터페이스를 구현하는 클래스는 JAR 파일의 META_INF/services/org.apache.flink.table.factories.Factory에 추가할 수 있습니다. 제공된 팩토리 식별자는 SQL CREATE CATALOG DDL 문에서 필요한 type 속성과의 일치에 사용됩니다.

Flink v1.16부터 TableEnvironment는 테이블 프로그램, SQL Client, SQL Gateway에서 일관된 클래스 로딩 동작을 위해 사용자 클래스 로더를 도입했습니다. 사용자 클래스로더는 ADD JAR 또는 CREATE FUNCTION .. USING JAR .. 문으로 추가된 jar 같은 모든 사용자 jar를 관리합니다. 사용자 정의 카탈로그는 클래스를 로드하기 위해 Thread.currentThread().getContextClassLoader()를 사용자 클래스 로더로 대체해야 합니다. 그렇지 않으면 ClassNotFoundException이 발생할 수 있습니다. 사용자 클래스 로더는 CatalogFactory.Context#getClassLoader를 통해 접근할 수 있습니다.

시간 여행(time travel)을 지원하기 위한 Catalog 인터페이스

버전 1.18부터 Flink 프레임워크는 테이블의 과거 데이터를 쿼리하기 위한 시간 여행을 지원합니다. 테이블의 과거 데이터를 쿼리하려면 테이블이 속한 카탈로그에 대해 getTable(ObjectPath tablePath, long timestamp) 메서드를 구현해야 합니다.

public class MyCatalogSupportTimeTravel implements Catalog {

    @Override
    public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp)
            throws TableNotExistException {
        // Build a schema corresponding to the specific time point.
        Schema schema = buildSchema(timestamp);
        // Set parameters to read data at the corresponding time point.
        Map<String, String> options = buildOptions(timestamp);
        // Build CatalogTable
        CatalogTable catalogTable =
                CatalogTable.newBuilder()
                        .schema(schema)
                        .comment("")
                        .partitionKeys(Collections.emptyList())
                        .options(options)
                        .snapshot(timestamp)
                        .build();
        return catalogTable;
    }
}

public class MyDynamicTableFactory implements DynamicTableSourceFactory {
    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        final ReadableConfig configuration =
                Configuration.fromMap(context.getCatalogTable().getOptions());

        // Get snapshot from CatalogTable
        final Optional<Long> snapshot = context.getCatalogTable().getSnapshot();

        // Build DynamicTableSource using snapshot options.
        final DynamicTableSource dynamicTableSource = buildDynamicSource(configuration, snapshot);

        return dynamicTableSource;
    }
}

Flink 테이블을 Catalog에 만들고 등록하는 방법 (How to Create and Register Flink Tables to Catalog)

SQL DDL 사용

사용자는 Table API와 SQL 양쪽에서 SQL DDL을 사용해 카탈로그에 테이블을 만들 수 있습니다.

Java

TableEnvironment tableEnv = ...;

// Create a HiveCatalog
Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>");

// Register the catalog
tableEnv.registerCatalog("myhive", catalog);

// Create a catalog database
tableEnv.executeSql("CREATE DATABASE mydb WITH (...)");

// Create a catalog table
tableEnv.executeSql("CREATE TABLE mytable (name STRING, age INT) WITH (...)");

tableEnv.listTables(); // should return the tables in current catalog and database.

Scala

val tableEnv = ...

// Create a HiveCatalog
val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")

// Register the catalog
tableEnv.registerCatalog("myhive", catalog)

// Create a catalog database
tableEnv.executeSql("CREATE DATABASE mydb WITH (...)")

// Create a catalog table
tableEnv.executeSql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")

tableEnv.listTables() // should return the tables in current catalog and database.

Python

from pyflink.table.catalog import HiveCatalog

# Create a HiveCatalog
catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")

# Register the catalog
t_env.register_catalog("myhive", catalog)

# Create a catalog database
t_env.execute_sql("CREATE DATABASE mydb WITH (...)")

# Create a catalog table
t_env.execute_sql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")

# should return the tables in current catalog and database.
t_env.list_tables()

SQL Client

// the catalog should have been registered via yaml file
Flink SQL> CREATE DATABASE mydb WITH (...);

Flink SQL> CREATE TABLE mytable (name STRING, age INT) WITH (...);

Flink SQL> SHOW TABLES;
mytable

자세한 정보는 Flink SQL CREATE DDL을 확인하세요.

Java, Scala 또는 Python 사용

사용자는 Java, Scala 또는 Python을 사용해 카탈로그 테이블을 프로그래밍 방식으로 만들 수 있습니다.

Java

import org.apache.flink.table.api.*;
import org.apache.flink.table.catalog.*;
import org.apache.flink.table.catalog.hive.HiveCatalog;

TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());

// Create a HiveCatalog
Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>");

// Register the catalog
tableEnv.registerCatalog("myhive", catalog);

// Create a catalog database
catalog.createDatabase("mydb", new CatalogDatabaseImpl(...));

// Create a catalog table
final Schema schema = Schema.newBuilder()
    .column("name", DataTypes.STRING())
    .column("age", DataTypes.INT())
    .build();

tableEnv.createTable("myhive.mydb.mytable", TableDescriptor.forConnector("kafka")
    .schema(schema)
    // …
    .build());

List<String> tables = catalog.listTables("mydb"); // tables should contain "mytable"

Scala

import org.apache.flink.table.api._
import org.apache.flink.table.catalog._
import org.apache.flink.table.catalog.hive.HiveCatalog

val tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode())

// Create a HiveCatalog
val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")

// Register the catalog
tableEnv.registerCatalog("myhive", catalog)

// Create a catalog database
catalog.createDatabase("mydb", new CatalogDatabaseImpl(...))

// Create a catalog table
val schema = Schema.newBuilder()
    .column("name", DataTypes.STRING())
    .column("age", DataTypes.INT())
    .build()

tableEnv.createTable("myhive.mydb.mytable", TableDescriptor.forConnector("kafka")
    .schema(schema)
    // …
    .build())

val tables = catalog.listTables("mydb") // tables should contain "mytable"

Python

from pyflink.table import *
from pyflink.table.catalog import HiveCatalog, CatalogDatabase, ObjectPath, CatalogBaseTable

settings = EnvironmentSettings.in_batch_mode()
t_env = TableEnvironment.create(settings)

# Create a HiveCatalog
catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")

# Register the catalog
t_env.register_catalog("myhive", catalog)

# Create a catalog database
database = CatalogDatabase.create_instance({"k1": "v1"}, None)
catalog.create_database("mydb", database)

# Create a catalog table
schema = Schema.new_builder() \
    .column("name", DataTypes.STRING()) \
    .column("age", DataTypes.INT()) \
    .build()

catalog_table = t_env.create_table("myhive.mydb.mytable", TableDescriptor.for_connector("kafka")
    .schema(schema)
    # …
    .build())

# tables should contain "mytable"
tables = catalog.list_tables("mydb")

Catalog API

참고: 여기에는 catalog 프로그램 API만 나열됩니다. 사용자는 SQL DDL로 많은 동일한 기능을 달성할 수 있습니다. 자세한 DDL 정보는 SQL CREATE DDL을 참조하세요.

데이터베이스 연산 (Database operations)

Java/Scala

// create database
catalog.createDatabase("mydb", new CatalogDatabaseImpl(...), false);

// drop database
catalog.dropDatabase("mydb", false);

// alter database
catalog.alterDatabase("mydb", new CatalogDatabaseImpl(...), false);

// get database
catalog.getDatabase("mydb");

// check if a database exist
catalog.databaseExists("mydb");

// list databases in a catalog
catalog.listDatabases();

Python

from pyflink.table.catalog import CatalogDatabase

# create database
catalog_database = CatalogDatabase.create_instance({"k1": "v1"}, None)
catalog.create_database("mydb", catalog_database, False)

# drop database
catalog.drop_database("mydb", False)

# alter database
catalog.alter_database("mydb", catalog_database, False)

# get database
catalog.get_database("mydb")

# check if a database exist
catalog.database_exists("mydb")

# list databases in a catalog
catalog.list_databases()

테이블 연산 (Table operations)

Java/Scala

// create table
catalog.createTable(new ObjectPath("mydb", "mytable"), CatalogTable.newBuilder()...build(), false);

// drop table
catalog.dropTable(new ObjectPath("mydb", "mytable"), false);

// alter table
catalog.alterTable(new ObjectPath("mydb", "mytable"), CatalogTable.newBuilder()...build(), false);

// rename table
catalog.renameTable(new ObjectPath("mydb", "mytable"), "my_new_table");

// get table
catalog.getTable("mytable");

// check if a table exist or not
catalog.tableExists("mytable");

// list tables in a database
catalog.listTables("mydb");

Python

from pyflink.table import *
from pyflink.table.catalog import CatalogBaseTable, ObjectPath
from pyflink.table.descriptors import Kafka

table_schema = TableSchema.builder() \
    .field("name", DataTypes.STRING()) \
    .field("age", DataTypes.INT()) \
    .build()

table_properties = Kafka() \
    .version("0.11") \
    .start_from_earlist() \
    .to_properties()

catalog_table = CatalogBaseTable.create_table(schema=table_schema, properties=table_properties, comment="my comment")

# create table
catalog.create_table(ObjectPath("mydb", "mytable"), catalog_table, False)

# drop table
catalog.drop_table(ObjectPath("mydb", "mytable"), False)

# alter table
catalog.alter_table(ObjectPath("mydb", "mytable"), catalog_table, False)

# rename table
catalog.rename_table(ObjectPath("mydb", "mytable"), "my_new_table")

# get table
catalog.get_table("mytable")

# check if a table exist or not
catalog.table_exists("mytable")

# list tables in a database
catalog.list_tables("mydb")

뷰 연산 (View operations)

Java/Scala

// create view
catalog.createTable(new ObjectPath("mydb", "myview"), new CatalogViewImpl(...), false);

// drop view
catalog.dropTable(new ObjectPath("mydb", "myview"), false);

// alter view
catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogViewImpl(...), false);

// rename view
catalog.renameTable(new ObjectPath("mydb", "myview"), "my_new_view", false);

// get view
catalog.getTable("myview");

// check if a view exist or not
catalog.tableExists("mytable");

// list views in a database
catalog.listViews("mydb");

Python

from pyflink.table import *
from pyflink.table.catalog import CatalogBaseTable, ObjectPath

table_schema = TableSchema.builder() \
    .field("name", DataTypes.STRING()) \
    .field("age", DataTypes.INT()) \
    .build()

catalog_table = CatalogBaseTable.create_view(
    original_query="select * from t1",
    expanded_query="select * from test-catalog.db1.t1",
    schema=table_schema,
    properties={},
    comment="This is a view"
)

catalog.create_table(ObjectPath("mydb", "myview"), catalog_table, False)

# drop view
catalog.drop_table(ObjectPath("mydb", "myview"), False)

# alter view
catalog.alter_table(ObjectPath("mydb", "mytable"), catalog_table, False)

# rename view
catalog.rename_table(ObjectPath("mydb", "myview"), "my_new_view", False)

# get view
catalog.get_table("myview")

# check if a view exist or not
catalog.table_exists("mytable")

# list views in a database
catalog.list_views("mydb")

파티션 연산 (Partition operations)

Java/Scala

// create view
catalog.createPartition(
    new ObjectPath("mydb", "mytable"),
    new CatalogPartitionSpec(...),
    new CatalogPartitionImpl(...),
    false);

// drop partition
catalog.dropPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...), false);

// alter partition
catalog.alterPartition(
    new ObjectPath("mydb", "mytable"),
    new CatalogPartitionSpec(...),
    new CatalogPartitionImpl(...),
    false);

// get partition
catalog.getPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));

// check if a partition exist or not
catalog.partitionExists(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));

// list partitions of a table
catalog.listPartitions(new ObjectPath("mydb", "mytable"));

// list partitions of a table under a give partition spec
catalog.listPartitions(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));

// list partitions of a table by expression filter
catalog.listPartitionsByFilter(new ObjectPath("mydb", "mytable"), Arrays.asList(epr1, ...));

Python

from pyflink.table.catalog import ObjectPath, CatalogPartitionSpec, CatalogPartition

catalog_partition = CatalogPartition.create_instance({}, "my partition")

catalog_partition_spec = CatalogPartitionSpec({"third": "2010", "second": "bob"})
catalog.create_partition(
    ObjectPath("mydb", "mytable"),
    catalog_partition_spec,
    catalog_partition,
    False)

# drop partition
catalog.drop_partition(ObjectPath("mydb", "mytable"), catalog_partition_spec, False)

# alter partition
catalog.alter_partition(
    ObjectPath("mydb", "mytable"),
    CatalogPartitionSpec(...),
    catalog_partition,
    False)

# get partition
catalog.get_partition(ObjectPath("mydb", "mytable"), catalog_partition_spec)

# check if a partition exist or not
catalog.partition_exists(ObjectPath("mydb", "mytable"), catalog_partition_spec)

# list partitions of a table
catalog.list_partitions(ObjectPath("mydb", "mytable"))

# list partitions of a table under a give partition spec
catalog.list_partitions(ObjectPath("mydb", "mytable"), catalog_partition_spec)

함수 연산 (Function operations)

Java/Scala

// create function
catalog.createFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);

// drop function
catalog.dropFunction(new ObjectPath("mydb", "myfunc"), false);

// alter function
catalog.alterFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);

// get function
catalog.getFunction("myfunc");

// check if a function exist or not
catalog.functionExists("myfunc");

// list functions in a database
catalog.listFunctions("mydb");

Python

from pyflink.table.catalog import ObjectPath, CatalogFunction

catalog_function = CatalogFunction.create_instance(class_name="my.python.udf")

# create function
catalog.create_function(ObjectPath("mydb", "myfunc"), catalog_function, False)

# drop function
catalog.drop_function(ObjectPath("mydb", "myfunc"), False)

# alter function
catalog.alter_function(ObjectPath("mydb", "myfunc"), catalog_function, False)

# get function
catalog.get_function("myfunc")

# check if a function exist or not
catalog.function_exists("myfunc")

# list functions in a database
catalog.list_functions("mydb")

Catalog용 Table API와 SQL (Table API and SQL for Catalog)

Catalog 등록 (Registering a Catalog)

사용자는 항상 기본적으로 생성되는 default_catalog라는 기본 인메모리 카탈로그에 접근할 수 있습니다. 이 카탈로그는 기본적으로 default_database라는 단일 데이터베이스를 가집니다. 사용자는 기존 Flink 세션에 추가 카탈로그를 등록할 수도 있습니다.

Java/Scala

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

Python

t_env.register_catalog(catalog)

YAML — YAML로 정의된 모든 카탈로그는 카탈로그의 유형을 지정하는 type 속성을 제공해야 합니다. 다음 유형이 기본적으로 지원됩니다.

Catalog	Type 값
GenericInMemory	generic_in_memory
Hive	hive

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

현재 Catalog와 데이터베이스 변경 (Changing the Current Catalog And Database)

Flink는 항상 현재 카탈로그와 데이터베이스에서 테이블, 뷰, UDF를 검색합니다.

Java/Scala

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

Python

t_env.use_catalog("myCatalog")
t_env.use_database("myDb")

SQL

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

현재 카탈로그가 아닌 카탈로그의 메타데이터는 catalog.database.object 형식의 완전히 정규화된 이름을 제공해 접근할 수 있습니다.

Java/Scala

tableEnv.from("not_the_current_catalog.not_the_current_db.my_table");

Python

t_env.from_path("not_the_current_catalog.not_the_current_db.my_table")

SQL

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

사용 가능한 Catalog 나열 (List Available Catalogs)

Java/Scala

tableEnv.listCatalogs();

Python

t_env.list_catalogs()

SQL

Flink SQL> show catalogs;

사용 가능한 데이터베이스 나열 (List Available Databases)

Java/Scala

tableEnv.listDatabases();

Python

t_env.list_databases()

SQL

Flink SQL> show databases;

사용 가능한 테이블 나열 (List Available Tables)

Java/Scala

tableEnv.listTables();

Python

t_env.list_tables()

SQL

Flink SQL> show tables;

Catalog 수정 리스너 (Catalog Modification Listener)

Flink는 데이터베이스와 테이블 ddl 같은 카탈로그 수정을 위한 사용자 정의 리스너 등록을 지원합니다. Flink는 ddl에 대해 CatalogModificationEvent 이벤트를 만들고 CatalogModificationListener에게 알립니다. 리스너를 구현하고 이벤트를 받을 때 외부 메타데이터 시스템에 정보를 보고하는 것 같은 사용자 정의 작업을 수행할 수 있습니다.

Catalog 리스너 구현 (Implement Catalog Listener)

카탈로그 수정 리스너에는 CatalogModificationListenerFactory(리스너 생성)와 CatalogModificationListener(이벤트 수신 및 처리) 두 가지 인터페이스가 있습니다. 이 인터페이스들을 구현해야 하며 아래는 예시입니다.

/** Factory used to create a {@link CatalogModificationListener} instance. */
public class YourCatalogListenerFactory implements CatalogModificationListenerFactory {
    /** The identifier for the customized listener factory, you can named it yourself. */
    private static final String IDENTIFIER = "your_factory";

    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }

    @Override
    public CatalogModificationListener createListener(Context context) {
        return new YourCatalogListener(Create http client from context);
    }
}

/** Customized catalog modification listener. */
public class YourCatalogListener implements CatalogModificationListener {
    private final HttpClient client;

    YourCatalogListener(HttpClient client) {
        this.client = client;
    }

    @Override
    public void onEvent(CatalogModificationEvent event) {
        // Report the database and table information via http client.
    }
}

사용자 정의 catalog listener factory를 위해 META-INF/services에 org.apache.flink.table.factories.Factory 파일을 만들고 내용을 the full name of YourCatalogListenerFactory로 지정해야 합니다. 그 후 코드를 jar 파일로 패키징하고 Flink 클러스터의 lib에 추가할 수 있습니다.

Catalog 리스너 등록 (Register Catalog Listener)

위 catalog modification factory와 listener를 구현한 후 테이블 환경에 등록할 수 있습니다.

Configuration configuration = new Configuration();

// Add the factory identifier, you can set multiple listeners in the configuration.
configuration.set(TableConfigOptions.TABLE_CATALOG_MODIFICATION_LISTENERS, Arrays.asList("your_factory"));
TableEnvironment env = TableEnvironment.create(
            EnvironmentSettings.newInstance()
                .withConfiguration(configuration)
                .build());

// Create/Alter/Drop database and table.
env.executeSql("CREATE TABLE ...").wait();

sql-gateway의 경우 Flink 구성 파일에 table.catalog-modification.listeners 옵션을 추가하고 게이트웨이를 시작하거나, 동적 파라미터로 sql-gateway를 시작한 다음 sql-client로 ddl을 직접 수행할 수 있습니다.

Catalog Store

Catalog Store는 카탈로그의 구성을 저장하는 데 사용됩니다. Catalog Store를 사용하면 세션에서 생성된 카탈로그의 구성이 Catalog Store의 해당 외부 시스템에 영속화됩니다. 세션이 재구성되더라도 이전에 생성된 카탈로그는 Catalog Store에서 검색할 수 있습니다.

Catalog Store 구성 (Configure Catalog Store)

사용자는 다양한 방법으로 Catalog Store를 구성할 수 있습니다. 하나는 Table API를 사용하는 것이고, 다른 하나는 YAML 구성을 사용하는 것입니다.

catalog store 인스턴스를 사용해 catalog store를 등록합니다.

// Initialize a catalog Store instance
CatalogStore catalogStore = new FileCatalogStore("file:///path/to/catalog/store/");

// set up the catalog store
final EnvironmentSettings settings =
        EnvironmentSettings.newInstance().inBatchMode()
        .withCatalogStore(catalogStore)
        .build();

구성을 사용해 catalog store를 등록합니다.

// Set up configuration
Configuration configuration = new Configuration();
configuration.set("table.catalog-store.kind", "file");
configuration.set("table.catalog-store.file.path", "file:///path/to/catalog/store/");
// set up the configuration.
final EnvironmentSettings settings =
        EnvironmentSettings.newInstance().inBatchMode()
        .withConfiguration(configuration)
        .build();

final TableEnvironment tableEnv = TableEnvironment.create(settings);

SQL Gateway에서는 모든 세션이 미리 생성된 Catalog를 자동으로 사용할 수 있도록 yaml 파일에 설정을 구성하는 것이 권장됩니다. 보통은 Catalog Store의 종류와 Catalog Store에 필요한 다른 필수 파라미터를 구성해야 합니다.

table.catalog-store.kind: file
table.catalog-store.file.path: file:///path/to/catalog/store/

Catalog Store 유형 (Catalog Store Type)

Flink에는 GenericInMemoryCatalogStore와 FileCatalogStore 두 가지 내장 Catalog Store가 있지만, Catalog Store 모델은 확장 가능하므로 사용자가 자신만의 사용자 정의 Catalog Store를 구현할 수도 있습니다.

GenericInMemoryCatalogStore

GenericInMemoryCatalogStore는 구성 정보를 메모리에 저장하는 CatalogStore의 구현입니다. 모든 카탈로그 구성은 세션의 수명 동안에만 사용할 수 있으며, 저장된 카탈로그 구성은 세션이 닫힌 후 자동으로 지워집니다.

기본적으로 Catalog Store 관련 설정이 지정되지 않으면 시스템이 이 구현을 사용합니다.

FileCatalogStore

FileCatalogStore는 Catalog 구성을 파일에 저장할 수 있습니다. FileCatalogStore를 사용하려면 Catalog 구성이 저장될 디렉터리를 지정해야 합니다. 각 Catalog는 Catalog 이름과 같은 이름의 자체 파일을 가집니다.

FileCatalogStore 구현은 Flink FileSystem 추상화를 통해 사용 가능한 로컬 및 원격 파일시스템을 모두 지원합니다. 주어진 Catalog Store 경로가 완전히 또는 부분적으로 존재하지 않으면 FileCatalogStore는 누락된 디렉터리를 만들려고 시도합니다.

주어진 Catalog Store 경로가 존재하지 않고 FileCatalogStore가 디렉터리를 만드는 데 실패하면 Catalog Store를 초기화할 수 없으므로 예외가 발생합니다. FileCatalogstore 초기화가 성공하지 못하면 SQL Client와 SQL Gateway 모두 손상됩니다.

다음은 FileCatalogStore를 사용한 Catalog 구성 저장을 나타내는 디렉터리 구조의 예시입니다.

- /path/to/save/the/catalog/
  - catalog1.yaml
  - catalog2.yaml
  - catalog3.yaml

Catalog Store 구성 (Catalog Store Configuration)

다음 옵션을 사용해 Catalog Store 동작을 조정할 수 있습니다.

키 (Key)	기본값 (Default)	타입 (Type)	설명 (Description)
table.catalog-store.kind	"generic_in_memory"	String	사용할 catalog store의 종류. 기본적으로 'generic_in_memory'와 'file' 옵션이 지원됩니다.
table.catalog-store.file.path	(none)	String	파일 catalog store 루트 디렉터리의 경로를 지정하기 위한 구성 옵션.

사용자 정의 Catalog Store (Custom Catalog Store)

Catalog Store는 확장 가능하며, 사용자는 인터페이스를 구현해 Catalog Store를 사용자 정의할 수 있습니다. SQL CLI나 SQL Gateway가 Catalog Store를 사용해야 한다면, 이 Catalog Store에 대해 해당 CatalogStoreFactory 인터페이스도 구현해야 합니다.

public class CustomCatalogStoreFactory implements CatalogStoreFactory {

    public static final String IDENTIFIER = "custom-kind";

    // Used to connect external storage systems
    private CustomClient client;

    @Override
    public CatalogStore createCatalogStore() {
        return new CustomCatalogStore();
    }

    @Override
    public void open(Context context) throws CatalogException {
        // initialize the resources, such as http client
        client = initClient(context);
    }

    @Override
    public void close() throws CatalogException {
        // release the resources
    }

    @Override
    public String factoryIdentifier() {
        // table store kind identifier
        return IDENTIFIER;
    }

    public Set<ConfigOption<?>> requiredOptions() {
        // define the required options
        Set<ConfigOption> options = new HashSet();
        options.add(OPTION_1);
        options.add(OPTION_2);

        return options;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        // define the optional options
    }
}

public class CustomCatalogStore extends AbstractCatalogStore {

    private Client client;

    public CustomCatalogStore(Client client) {
        this.client = client;
    }

    @Override
    public void storeCatalog(String catalogName, CatalogDescriptor catalog)
            throws CatalogException {
        // store the catalog
    }

    @Override
    public void removeCatalog(String catalogName, boolean ignoreIfNotExists)
            throws CatalogException {
        // remove the catalog descriptor
    }

    @Override
    public Optional<CatalogDescriptor> getCatalog(String catalogName) {
        // retrieve the catalog configuration and build the catalog descriptor
    }

    @Override
    public Set<String> listCatalogs() {
        // list all catalogs
    }

    @Override
    public boolean contains(String catalogName) {
    }
}

더 알아보기 (Learn more)

• Flink SQL CREATE DDL
• Hive Catalog
• JdbcCatalog
• Flink SQL 쿼리