Catalogs

Catalogs provide metadata, such as databases, tables, partitions, views, and functions and information needed to access data stored in a database or other external systems.

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.

Catalog Types

GenericInMemoryCatalog

The GenericInMemoryCatalog 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 JdbcCatalog documentation 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 Hive documentation provides full details on setting up the catalog and interfacing with an existing Hive installation.

Warning The Hive Metastore stores all meta-object names in lower case. This is unlike GenericInMemoryCatalog which is case-sensitive

User-Defined Catalog

Catalogs are pluggable and users can develop custom catalogs by implementing the Catalog interface. To use custom catalogs in SQL CLI, users should develop both a catalog and its corresponding catalog factory by implementing the CatalogFactory interface.

The catalog factory defines a set of properties for configuring the catalog when the SQL CLI bootstraps. The set of properties will be passed to a discovery service where the service tries to match the properties to a CatalogFactory and initiate a corresponding catalog instance.

How to Create and Register Flink Tables to Catalog

Using SQL DDL

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

  1. TableEnvironment tableEnv = ...
  3. // Create a HiveCatalog 
  4. Catalog 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.
  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.
  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()
  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.

Using Java, Scala or Python

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

  1. import org.apache.flink.table.api.*;
  2. import org.apache.flink.table.catalog.*;
  3. import org.apache.flink.table.catalog.hive.HiveCatalog;
  4. import org.apache.flink.table.descriptors.Kafka;
  6. TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());
  8. // Create a HiveCatalog 
  9. Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>");
  11. // Register the catalog
  12. tableEnv.registerCatalog("myhive", catalog);
  14. // Create a catalog database 
  15. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...));
  17. // Create a catalog table
  18. TableSchema schema = TableSchema.builder()
  19.     .field("name", DataTypes.STRING())
  20.     .field("age", DataTypes.INT())
  21.     .build();
  23. catalog.createTable(
  24.         new ObjectPath("mydb", "mytable"), 
  25.         new CatalogTableImpl(
  26.             schema,
  27.             new Kafka()
  28.                 .version("0.11")
  29.                 ....
  30.                 .startFromEarlist()
  31.                 .toProperties(),
  32.             "my comment"
  33.         ),
  34.         false
  35.     );
  37. 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
  4. import org.apache.flink.table.descriptors.Kafka
  6. val tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance.build)
  8. // Create a HiveCatalog 
  9. val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")
  11. // Register the catalog
  12. tableEnv.registerCatalog("myhive", catalog)
  14. // Create a catalog database 
  15. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...))
  17. // Create a catalog table
  18. val schema = TableSchema.builder()
  19.     .field("name", DataTypes.STRING())
  20.     .field("age", DataTypes.INT())
  21.     .build()
  23. catalog.createTable(
  24.         new ObjectPath("mydb", "mytable"), 
  25.         new CatalogTableImpl(
  26.             schema,
  27.             new Kafka()
  28.                 .version("0.11")
  29.                 ....
  30.                 .startFromEarlist()
  31.                 .toProperties(),
  32.             "my comment"
  33.         ),
  34.         false
  35.     )
  37. val tables = catalog.listTables("mydb") // tables should contain "mytable"
  1. from pyflink.table import *
  2. from pyflink.table.catalog import HiveCatalog, CatalogDatabase, ObjectPath, CatalogBaseTable
  3. from pyflink.table.descriptors import Kafka
  5. settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()
  6. t_env = BatchTableEnvironment.create(environment_settings=settings)
  8. # Create a HiveCatalog
  9. catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")
  11. # Register the catalog
  12. t_env.register_catalog("myhive", catalog)
  14. # Create a catalog database
  15. database = CatalogDatabase.create_instance({"k1": "v1"}, None)
  16. catalog.create_database("mydb", database)
  18. # Create a catalog table
  19. table_schema = TableSchema.builder() \
  20.     .field("name", DataTypes.STRING()) \
  21.     .field("age", DataTypes.INT()) \
  22.     .build()
  24. table_properties = Kafka() \
  25.     .version("0.11") \
  26.     .start_from_earlist() \
  27.     .to_properties()
  29. catalog_table = CatalogBaseTable.create_table(
  30.     schema=table_schema, properties=table_properties, comment="my comment")
  32. catalog.create_table(
  33.     ObjectPath("mydb", "mytable"),
  34.     catalog_table,
  35.     False)
  37. # tables should contain "mytable"
  38. tables = catalog.list_tables("mydb")

Catalog API

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

Database operations

  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();
  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

  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");
  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

  1. // create view
  2. catalog.createTable(new ObjectPath("mydb", "myview"), new CatalogViewImpl(...), false);
  4. // drop view
  5. catalog.dropTable(new ObjectPath("mydb", "myview"), false);
  7. // alter view
  8. catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogViewImpl(...), false);
  10. // rename view
  11. catalog.renameTable(new ObjectPath("mydb", "myview"), "my_new_view", false);
  13. // get view
  14. catalog.getTable("myview");
  16. // check if a view exist or not
  17. catalog.tableExists("mytable");
  19. // list views in a database
  20. catalog.listViews("mydb");
  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
  26. catalog.rename_table(ObjectPath("mydb", "myview"), "my_new_view", False)
  28. # get view
  29. catalog.get_table("myview")
  31. # check if a view exist or not
  32. catalog.table_exists("mytable")
  34. # list views in a database
  35. catalog.list_views("mydb")

Partition operations

  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, ...));
  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)

Function operations

  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");
  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")

Table API and SQL for Catalog

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.

  1. tableEnv.registerCatalog(new CustomCatalog("myCatalog"));
  1. t_env.register_catalog(catalog)

All catalogs defined using YAML must provide a type property that specifies the type of catalog. The following types are supported out of the box.

CatalogType Value
GenericInMemorygeneric_in_memory
Hivehive
  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.

  1. tableEnv.useCatalog("myCatalog");
  2. tableEnv.useDatabase("myDb");
  1. t_env.use_catalog("myCatalog")
  2. t_env.use_database("myDb")
  1. Flink SQL> USE CATALOG myCatalog;
  2. Flink SQL> USE myDB;

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

  1. tableEnv.from("not_the_current_catalog.not_the_current_db.my_table");
  1. t_env.from_path("not_the_current_catalog.not_the_current_db.my_table")
  1. Flink SQL> SELECT * FROM not_the_current_catalog.not_the_current_db.my_table;

List Available Catalogs

  1. tableEnv.listCatalogs();
  1. t_env.list_catalogs()
  1. Flink SQL> show catalogs;

List Available Databases

  1. tableEnv.listDatabases();
  1. t_env.list_databases()
  1. Flink SQL> show databases;

List Available Tables

  1. tableEnv.listTables();
  1. t_env.list_tables()
  1. Flink SQL> show tables;