Catalogs

    One of the most crucial aspects of data processing is managing metadata. It may be transient metadata like temporary tables, or UDFs registered against the table environment. Or permanent metadata, like that in a Hive Metastore. Catalogs provide a unified API for managing metadata and making it accessible from the Table API and SQL Queries.

    Catalog enables users to reference existing metadata in their data systems, and automatically maps them to Flink’s corresponding metadata. For example, Flink can map JDBC tables to Flink table automatically, and users don’t have to manually re-writing DDLs in Flink. Catalog greatly simplifies steps required to get started with Flink with users’ existing system, and greatly enhanced user experiences.

    The is an in-memory implementation of a catalog. All objects will be available only for the lifetime of the session.

    JdbcCatalog

    The JdbcCatalog enables users to connect Flink to relational databases over JDBC protocol. PostgresCatalog is the only implementation of JDBC Catalog at the moment. See for more details on setting up the catalog.

    HiveCatalog

    The HiveCatalog serves two purposes; as persistent storage for pure Flink metadata, and as an interface for reading and writing existing Hive metadata. Flink’s provides full details on setting up the catalog and interfacing with an existing Hive installation.

    User-Defined Catalog

    Catalogs are pluggable and users can develop custom catalogs by implementing the Catalog interface.

    In order to use custom catalogs with Flink SQL, users should implement a corresponding catalog factory by implementing the CatalogFactory interface. The factory is discovered using Java’s Service Provider Interfaces (SPI). Classes that implement this interface can be added to META_INF/services/org.apache.flink.table.factories.Factory in JAR files. The provided factory identifier will be used for matching against the required type property in a SQL CREATE CATALOG DDL statement.

    Using SQL DDL

    Users can use SQL DDL to create tables in catalogs in both Table API and SQL.

    Java

    Scala

    1. val tableEnv = ...
    3. // Create a HiveCatalog 
    4. val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")
    6. // Register the catalog
    7. tableEnv.registerCatalog("myhive", catalog)
    9. // Create a catalog database
    10. tableEnv.executeSql("CREATE DATABASE mydb WITH (...)")
    12. // Create a catalog table
    13. tableEnv.executeSql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")
    15. tableEnv.listTables() // should return the tables in current catalog and database.

    Python

    1. from pyflink.table.catalog import HiveCatalog
    3. # Create a HiveCatalog
    4. catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")
    6. # Register the catalog
    7. t_env.register_catalog("myhive", catalog)
    9. # Create a catalog database
    10. t_env.execute_sql("CREATE DATABASE mydb WITH (...)")
    12. # Create a catalog table
    13. t_env.execute_sql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")
    15. # should return the tables in current catalog and database.
    16. t_env.list_tables()

    SQL Client

    1. // the catalog should have been registered via yaml file
    2. Flink SQL> CREATE DATABASE mydb WITH (...);
    4. Flink SQL> CREATE TABLE mytable (name STRING, age INT) WITH (...);
    6. Flink SQL> SHOW TABLES;
    7. mytable

    For detailed information, please check out Flink SQL CREATE DDL.

    Users can use Java, Scala or Python to create catalog tables programmatically.

    Java

    1. import org.apache.flink.table.api.*;
    2. import org.apache.flink.table.catalog.*;
    3. import org.apache.flink.table.catalog.hive.HiveCatalog;
    5. TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
    7. // Create a HiveCatalog 
    8. Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>");
    10. // Register the catalog
    11. tableEnv.registerCatalog("myhive", catalog);
    13. // Create a catalog database 
    14. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...));
    16. // Create a catalog table
    17. final Schema schema = Schema.newBuilder()
    18.     .column("name", DataTypes.STRING())
    19.     .column("age", DataTypes.INT())
    20.     .build();
    22. tableEnv.createTable("myhive.mydb.mytable", TableDescriptor.forConnector("kafka")
    23.     .schema(schema)
    24.     // …
    25.     .build());
    27. List<String> tables = catalog.listTables("mydb"); // tables should contain "mytable"
    1. import org.apache.flink.table.api._
    2. import org.apache.flink.table.catalog._
    3. import org.apache.flink.table.catalog.hive.HiveCatalog
    5. val tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode())
    7. // Create a HiveCatalog 
    8. val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")
    10. // Register the catalog
    11. tableEnv.registerCatalog("myhive", catalog)
    13. // Create a catalog database 
    14. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...))
    16. // Create a catalog table
    17. val schema = Schema.newBuilder()
    18.     .column("name", DataTypes.STRING())
    19.     .column("age", DataTypes.INT())
    20.     .build()
    22. tableEnv.createTable("myhive.mydb.mytable", TableDescriptor.forConnector("kafka")
    23.     .schema(schema)
    24.     // …
    25.     .build())
    27. val tables = catalog.listTables("mydb") // tables should contain "mytable"

    Python

    1. from pyflink.table import *
    2. from pyflink.table.catalog import HiveCatalog, CatalogDatabase, ObjectPath, CatalogBaseTable
    4. settings = EnvironmentSettings.in_batch_mode()
    5. t_env = TableEnvironment.create(settings)
    7. # Create a HiveCatalog
    8. catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")
    10. # Register the catalog
    11. t_env.register_catalog("myhive", catalog)
    13. # Create a catalog database
    14. database = CatalogDatabase.create_instance({"k1": "v1"}, None)
    15. catalog.create_database("mydb", database)
    17. # Create a catalog table
    18.     .column("name", DataTypes.STRING()) \
    19.     .column("age", DataTypes.INT()) \
    20.     .build()
    22. catalog_table = t_env.create_table("myhive.mydb.mytable", TableDescriptor.for_connector("kafka")
    23.     .schema(schema)
    24.     // …
    25.     .build())
    27. # tables should contain "mytable"
    28. tables = catalog.list_tables("mydb")

    Note: only catalog program APIs are listed here. Users can achieve many of the same functionalities with SQL DDL. For detailed DDL information, please refer to .

    Database operations

    Java/Scala

    1. // create database
    2. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...), false);
    4. // drop database
    5. catalog.dropDatabase("mydb", false);
    7. // alter database
    8. catalog.alterDatabase("mydb", new CatalogDatabaseImpl(...), false);
    10. // get database
    11. catalog.getDatabase("mydb");
    13. // check if a database exist
    14. catalog.databaseExists("mydb");
    16. // list databases in a catalog
    17. catalog.listDatabases();

    Python

    1. from pyflink.table.catalog import CatalogDatabase
    3. # create database
    4. catalog_database = CatalogDatabase.create_instance({"k1": "v1"}, None)
    5. catalog.create_database("mydb", catalog_database, False)
    7. # drop database
    8. catalog.drop_database("mydb", False)
    10. # alter database
    11. catalog.alter_database("mydb", catalog_database, False)
    13. # get database
    14. catalog.get_database("mydb")
    16. # check if a database exist
    17. catalog.database_exists("mydb")
    19. # list databases in a catalog
    20. catalog.list_databases()

    Table operations

    Java/Scala

    1. // create table
    2. catalog.createTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
    4. // drop table
    5. catalog.dropTable(new ObjectPath("mydb", "mytable"), false);
    7. // alter table
    8. catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
    10. // rename table
    11. catalog.renameTable(new ObjectPath("mydb", "mytable"), "my_new_table");
    13. // get table
    14. catalog.getTable("mytable");
    16. // check if a table exist or not
    17. catalog.tableExists("mytable");
    19. // list tables in a database
    20. catalog.listTables("mydb");

    Python

    1. from pyflink.table import *
    2. from pyflink.table.catalog import CatalogBaseTable, ObjectPath
    3. from pyflink.table.descriptors import Kafka
    5. table_schema = TableSchema.builder() \
    6.     .field("name", DataTypes.STRING()) \
    7.     .field("age", DataTypes.INT()) \
    8.     .build()
    10. table_properties = Kafka() \
    11.     .version("0.11") \
    12.     .start_from_earlist() \
    13.     .to_properties()
    15. catalog_table = CatalogBaseTable.create_table(schema=table_schema, properties=table_properties, comment="my comment")
    17. # create table
    18. catalog.create_table(ObjectPath("mydb", "mytable"), catalog_table, False)
    20. # drop table
    21. catalog.drop_table(ObjectPath("mydb", "mytable"), False)
    23. # alter table
    24. catalog.alter_table(ObjectPath("mydb", "mytable"), catalog_table, False)
    26. # rename table
    27. catalog.rename_table(ObjectPath("mydb", "mytable"), "my_new_table")
    29. # get table
    30. catalog.get_table("mytable")
    32. # check if a table exist or not
    33. catalog.table_exists("mytable")
    35. # list tables in a database
    36. catalog.list_tables("mydb")

    View operations

    Java/Scala

    Python

    1. from pyflink.table import *
    2. from pyflink.table.catalog import CatalogBaseTable, ObjectPath
    4. table_schema = TableSchema.builder() \
    5.     .field("name", DataTypes.STRING()) \
    6.     .field("age", DataTypes.INT()) \
    7.     .build()
    9. catalog_table = CatalogBaseTable.create_view(
    10.     original_query="select * from t1",
    11.     expanded_query="select * from test-catalog.db1.t1",
    12.     schema=table_schema,
    13.     properties={},
    14.     comment="This is a view"
    15. )
    17. catalog.create_table(ObjectPath("mydb", "myview"), catalog_table, False)
    19. # drop view
    20. catalog.drop_table(ObjectPath("mydb", "myview"), False)
    22. # alter view
    23. catalog.alter_table(ObjectPath("mydb", "mytable"), catalog_table, False)
    25. # rename view
    27. # get view
    28. catalog.get_table("myview")
    30. # check if a view exist or not
    31. catalog.table_exists("mytable")
    33. # list views in a database
    34. catalog.list_views("mydb")

    Partition operations

    Java/Scala

    1. // create view
    2. catalog.createPartition(
    3.     new ObjectPath("mydb", "mytable"),
    4.     new CatalogPartitionSpec(...),
    5.     new CatalogPartitionImpl(...),
    6.     false);
    8. // drop partition
    9. catalog.dropPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...), false);
    11. // alter partition
    12. catalog.alterPartition(
    13.     new ObjectPath("mydb", "mytable"),
    14.     new CatalogPartitionSpec(...),
    15.     new CatalogPartitionImpl(...),
    16.     false);
    18. // get partition
    19. catalog.getPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));
    21. // check if a partition exist or not
    22. catalog.partitionExists(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));
    24. // list partitions of a table
    25. catalog.listPartitions(new ObjectPath("mydb", "mytable"));
    27. // list partitions of a table under a give partition spec
    28. catalog.listPartitions(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));
    30. // list partitions of a table by expression filter
    31. catalog.listPartitionsByFilter(new ObjectPath("mydb", "mytable"), Arrays.asList(epr1, ...));

    Python

    1. from pyflink.table.catalog import ObjectPath, CatalogPartitionSpec, CatalogPartition
    3. catalog_partition = CatalogPartition.create_instance({}, "my partition")
    5. catalog_partition_spec = CatalogPartitionSpec({"third": "2010", "second": "bob"})
    6. catalog.create_partition(
    7.     ObjectPath("mydb", "mytable"),
    8.     catalog_partition_spec,
    9.     catalog_partition,
    10.     False)
    12. # drop partition
    13. catalog.drop_partition(ObjectPath("mydb", "mytable"), catalog_partition_spec, False)
    15. # alter partition
    16. catalog.alter_partition(
    17.     ObjectPath("mydb", "mytable"),
    18.     CatalogPartitionSpec(...),
    19.     catalog_partition,
    20.     False)
    22. # get partition
    23. catalog.get_partition(ObjectPath("mydb", "mytable"), catalog_partition_spec)
    25. # check if a partition exist or not
    26. catalog.partition_exists(ObjectPath("mydb", "mytable"), catalog_partition_spec)
    28. # list partitions of a table
    29. catalog.list_partitions(ObjectPath("mydb", "mytable"))
    31. # list partitions of a table under a give partition spec
    32. catalog.list_partitions(ObjectPath("mydb", "mytable"), catalog_partition_spec)

    Java/Scala

    1. // create function
    2. catalog.createFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);
    4. // drop function
    5. catalog.dropFunction(new ObjectPath("mydb", "myfunc"), false);
    7. // alter function
    8. catalog.alterFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);
    10. // get function
    11. catalog.getFunction("myfunc");
    13. // check if a function exist or not
    14. catalog.functionExists("myfunc");
    16. // list functions in a database
    17. catalog.listFunctions("mydb");

    Python

    1. from pyflink.table.catalog import ObjectPath, CatalogFunction
    3. catalog_function = CatalogFunction.create_instance(class_name="my.python.udf")
    5. # create function
    6. catalog.create_function(ObjectPath("mydb", "myfunc"), catalog_function, False)
    8. # drop function
    9. catalog.drop_function(ObjectPath("mydb", "myfunc"), False)
    11. # alter function
    12. catalog.alter_function(ObjectPath("mydb", "myfunc"), catalog_function, False)
    14. # get function
    15. catalog.get_function("myfunc")
    17. # check if a function exist or not
    18. catalog.function_exists("myfunc")
    20. # list functions in a database
    21. catalog.list_functions("mydb")

    Registering a Catalog

    Users have access to a default in-memory catalog named default_catalog, that is always created by default. This catalog by default has a single database called default_database. Users can also register additional catalogs into an existing Flink session.

    Java/Scala

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

    Python

    1. t_env.register_catalog(catalog)

    YAML

    1. catalogs:
    2.    - name: myCatalog
    3.      type: custom_catalog
    4.      hive-conf-dir: ...

    Changing the Current Catalog And Database

    Flink will always search for tables, views, and UDF’s in the current catalog and database.

    Java/Scala

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

    Python

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

    SQL

    Metadata from catalogs that are not the current catalog are accessible by providing fully qualified names in the form catalog.database.object.

    Java/Scala

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

    Python

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

    SQL

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

    List Available Catalogs

    Java/Scala

    1. tableEnv.listCatalogs();

    Python

    1. t_env.list_catalogs()

    SQL

    1. Flink SQL> show catalogs;

    List Available Databases

    Java/Scala

    1. tableEnv.listDatabases();

    Python

    1. t_env.list_databases()

    SQL

    1. Flink SQL> show databases;

    Java/Scala

      Python