Confluent Developer Marketplace Get started
  1. Home
  2. Configure Kubernetes Clusters, Environments, Catalogs, Compute Pools, Artifacts, and Access Control for Confluent Platform for Apache Flink

Manage Environment Catalogs in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) provides an environment catalog for every environment. The environment catalog is a writable Flink SQL catalog that stores metadata you create with Data Definition Language (DDL) statements, such as user-defined functions, tables that use custom connectors, and views.

The built-in Kafka catalogs cannot store this metadata. The catalogs expose Kafka topics as tables and derive their schemas from Schema Registry, but they do not accept CREATE FUNCTION or CREATE VIEW, and they cannot store tables that are backed by systems other than Kafka. The environment catalog fills that gap. You register functions, custom-connector tables, and views in it. Because the catalog is scoped to a single environment, the objects you create are visible only within that environment.

Note

The environment catalog stores metadata only. You deliver the function and connector code by uploading it to artifact management and referencing it from your DDL.

How it works

When the environment catalog is enabled, CMF adds one catalog to every environment automatically. You do not create or delete it.

  • The catalog name is _env_ followed by the environment name. For example, environment prod has the catalog _env_prod. The _env_ prefix keeps the name from colliding with your Kafka catalogs.

  • Each environment catalog includes a default database that CMF creates automatically and that you cannot delete. CMF creates it the first time a statement in the environment opens the catalog, not when the environment is created. You can create additional databases with CREATE DATABASE and drop them with DROP DATABASE.

  • CMF stores the catalog’s objects (functions, tables, and views) in its own database. No Schema Registry or Kafka cluster is required.

The objects you can create in the environment catalog are:

Enable the environment catalog

The environment catalog is an opt-in feature and is disabled by default. Enable it in your Helm values file:

cmf:
  sql:
    environmentCatalog:
      enabled: true

While the feature is disabled, the environment catalog is not registered, and the catalog endpoint returns HTTP 404.

To register and use functions or custom connectors, also enable artifact management, which stores the JAR files that your DDL references. See User-Defined Functions in Confluent Manager for Apache Flink and Custom Connectors and Formats in Confluent Manager for Apache Flink. With both features enabled, all statements of environments that have artifacts compile on compile workers. See Compile workers.

Important

The environment catalog runtime is included in the cp-flink-sql image starting with version 1.19-cp9. Compute pools must run confluentinc/cp-flink-sql:1.19-cp9 or later. Statements that use the environment catalog fail on earlier images, such as 1.19-cp8, even when you enable the feature in CMF. The recommended image is confluentinc/cp-flink-sql:1.19-cp11, which includes the latest bug fixes validated with this CMF release. Set the pool’s image field accordingly. See Manage Compute Pools in Confluent Manager for Apache Flink.

Use the environment catalog

To create or use objects in the environment catalog, set the statement’s current catalog and database to the environment catalog and its default database.

With the Confluent CLI, pass the --catalog and --database flags:

confluent flink statement create show-functions \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool \
  --catalog _env_prod \
  --database default \
  --sql "SHOW USER FUNCTIONS;"

With the REST API, set the current catalog and database in the statement’s properties:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Statement",
  "metadata": {
    "name": "show-functions"
  },
  "spec": {
    "statement": "SHOW USER FUNCTIONS;",
    "computePoolName": "pool",
    "properties": {
      "sql.current-catalog": "_env_prod",
      "sql.current-database": "default"
    }
  }
}

Within the environment catalog you can run function DDL (CREATE FUNCTION, DROP FUNCTION), table DDL (CREATE TABLE, ALTER TABLE, DROP TABLE), view DDL (CREATE VIEW, ALTER VIEW, DROP VIEW), database DDL (CREATE DATABASE, DROP DATABASE), and queries (SELECT, INSERT INTO) that reference those objects.

Inspect the environment catalog with the REST API

CMF exposes a read-only description of an environment catalog and its databases through the REST API:

curl -v -X GET http://cmf:8080/cmf/api/v1/environments/prod/catalog

The response contains the catalog name and its databases. CMF creates the default database on first use. Until a statement has run in the environment, the list of databases can be empty. The response does not list the functions, tables, or views in the catalog. To list them, run a SHOW USER FUNCTIONS;, SHOW TABLES;, or SHOW VIEWS; statement against the catalog. The endpoint returns HTTP 404 when the environment catalog feature is disabled.

Limitations

  • CREATE TABLE AS (CTAS) and CREATE OR REPLACE TABLE AS (RTAS) statements are not supported in the environment catalog.

  • UPDATE and DELETE statements are not supported, in line with Flink SQL, which does not support these DML statements in streaming mode. Use INSERT INTO to write rows and SELECT to read them.

  • CREATE TEMPORARY FUNCTION and CREATE TEMPORARY SYSTEM FUNCTION are not supported. Register functions with CREATE FUNCTION instead.

  • The environment catalog and its default database cannot be renamed or deleted.

  • User-defined functions must be implemented in Java. See User-Defined Functions in Confluent Manager for Apache Flink.

  • You can’t delete an environment while the default database of its environment catalog holds functions, tables, or views, or while the catalog holds databases you created. Drop them first. For more information, see Configure Environments in Confluent Manager for Apache Flink.

Related content