Realtime Compute for Apache Flink:Manage Kafka JSON catalogs

What is a Kafka JSON catalog?

A Kafka JSON catalog automatically infers table structures from JSON-formatted Kafka messages, letting you query topics with SQL instead of writing DDL statements.

Key benefits:

• Faster development: Query Kafka topics directly without declaring schemas

• Fewer errors: Table names automatically match topic names

• Schema evolution: Use with CTAS to sync data when schemas change

Example: Instead of writing this DDL statement:

CREATE TABLE orders (
  order_id STRING,
  product_name STRING,
  quantity INT,
  price DECIMAL(10,2)
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = '...',
  'format' = 'json'
);

You can directly query the topic:

SELECT * FROM kafka_catalog.kafka.orders;

Limitations

• Only JSON-formatted messages are supported.

• Requires VVR 6.0.2 or later.

• Modifying a Kafka JSON catalog is not supported.

• Catalog tables are read-only.

  Note

  In CREATE DATABASE AS (CDAS) or CREATE TABLE AS (CTAS) scenarios that use a Kafka JSON catalog, topics can be created automatically.

• Kafka JSON catalogs cannot read from or write to Kafka clusters that have SSL or SASL authentication enabled.

• Tables provided by Kafka JSON catalogs can be used directly as source tables in Flink SQL jobs. They cannot be used as sink tables or lookup dimension tables.

• ApsaraMQ for Kafka currently does not allow you to delete consumer groups using the same API operation as Apache Kafka. When you create a Kafka JSON catalog, you must configure the aliyun.kafka.instanceId, aliyun.kafka.accessKeyId, aliyun.kafka.accessKeySecret, aliyun.kafka.endpoint, and aliyun.kafka.regionId parameters to automatically delete consumer groups. Comparison between ApsaraMQ for Kafka and Apache Kafka.

Usage notes

Schema consistency: The catalog samples messages to infer schemas. If messages have different formats, the catalog merges all fields into one schema.

Impact of schema changes: If the message format changes, job restarts may fail because the execution plan uses the old schema.

Solution: Fix the schema with CREATE TEMPORARY TABLE. Example:

-- Define a fixed schema based on the catalog table
CREATE TEMPORARY TABLE orders (
  value_order_id STRING,
  value_product_name STRING,
  value_quantity INT,
  value_price DECIMAL(10,2)
) LIKE `kafka_catalog`.`kafka`.`orders`;

This locks the schema, preventing restart failures.

Create a Kafka JSON catalog

1. In the SQL editor on Scripts, enter the statement to create a Kafka JSON catalog.

   • Self-managed Kafka cluster or EMR on ECS Kafka cluster

     CREATE CATALOG <YourCatalogName> WITH(
      'type'='kafka',  -- Required
      'properties.bootstrap.servers'='<brokers>',  -- Required
      'format'='json',  -- Required
      'default-database'='<dbName>',
      'key.fields-prefix'='<keyPrefix>',
      'value.fields-prefix'='<valuePrefix>',
      'timestamp-format.standard'='<timestampFormat>',
      'infer-schema.flatten-nested-columns.enable'='<flattenNestedColumns>',
      'infer-schema.primitive-as-string'='<primitiveAsString>',
      'infer-schema.parse-key-error.field-name'='<parseKeyErrorFieldName>',
      'infer-schema.compacted-topic-as-upsert-table'='true',
      'max.fetch.records'='100'
     );

   • ApsaraMQ for Kafka

     CREATE CATALOG <YourCatalogName> WITH(
      'type'='kafka',  -- Required
      'properties.bootstrap.servers'='<brokers>',  -- Required
      'format'='json',  -- Required
      'default-database'='<dbName>',
      'key.fields-prefix'='<keyPrefix>',
      'value.fields-prefix'='<valuePrefix>',
      'timestamp-format.standard'='<timestampFormat>',
      'infer-schema.flatten-nested-columns.enable'='<flattenNestedColumns>',
      'infer-schema.primitive-as-string'='<primitiveAsString>',
      'infer-schema.parse-key-error.field-name'='<parseKeyErrorFieldName>',
      'infer-schema.compacted-topic-as-upsert-table'='true',
      'max.fetch.records'='100',
      'aliyun.kafka.accessKeyId'='<aliyunAccessKeyId>',  -- Required
      'aliyun.kafka.accessKeySecret'='<aliyunAccessKeySecret>',  -- Required
      'aliyun.kafka.instanceId'='<aliyunKafkaInstanceId>',  -- Required
      'aliyun.kafka.endpoint'='<aliyunKafkaEndpoint>',  -- Required
      'aliyun.kafka.regionId'='<aliyunKafkaRegionId>'  -- Required
     );

   Parameter	Type	Description	Required	Remarks
   YourCatalogName	String	Catalog name.	Yes	Enter a custom name. Important Remove the angle brackets (<>) when you replace the placeholder.
   type	String	Catalog type.	Yes	Must be kafka.
   properties.bootstrap.servers	String	Kafka broker addresses.	Yes	Format: host1:port1,host2:port2,host3:port3. Separate multiple addresses with commas (,).
   format	String	Kafka message format.	Yes	Must be json.
   default-database	String	Cluster name.	No	Default: kafka. Sets db_name in the three-part name catalog_name.db_name.table_name. Kafka has no databases, so use any string.
   key.fields-prefix	String	Prefix for message key fields. Avoids naming conflicts.	No	Default: key_. Example: Field a becomes key_a. Note The value of the key.fields-prefix parameter cannot be a prefix of the value of the value.fields-prefix parameter. For example, if you set value.fields-prefix to test1_value_, you cannot set key.fields-prefix to test1_.
   value.fields-prefix	String	Prefix for message value fields. Avoids naming conflicts.	No	Default: value_. Example: Field b becomes value_b. Note The value of the value.fields-prefix parameter cannot be a prefix of the value of the key.fields-prefix parameter. For example, if you set key.fields-prefix to test2_value_, you cannot set value.fields-prefix to test2_.
   timestamp-format.standard	String	Timestamp format for JSON messages. Flink tries the configured format first, then falls back to other formats.	No	Valid values: SQL (default) ISO-8601
   infer-schema.flatten-nested-columns.enable	Boolean	Recursively expand nested columns in message values.	No	Valid values: true: Expand nested columns. Expanded column names use the path as the name. Example: col in {"nested": {"col": true}} becomes nested.col. Note If you set this parameter to true, use it with the CREATE TABLE AS (CTAS) statement. Other DML statements do not support automatic expansion of nested columns. false (default): Treat nested types as String.
   infer-schema.primitive-as-string	Boolean	Infer all primitive types as String.	No	Valid values: true: Infer all primitive types as String. false (default): Infer types based on basic rules.
   infer-schema.parse-key-error.field-name	String	If the message key is non-empty but unparseable, a VARBINARY field is added. The field name combines the key.fields-prefix prefix with this parameter value.	No	Default: col. Example: If the value parses to value_name and the key fails to parse, the schema contains key_col and value_name.
   infer-schema.compacted-topic-as-upsert-table	Boolean	Treat the table as an Upsert Kafka table when the topic cleanup policy is compact and the message key is non-empty.	No	Default: true. Enable when using CTAS or CDAS to sync data to ApsaraMQ for Kafka. Note Only VVR 6.0.2 or later support this parameter.
   max.fetch.records	Int	Maximum messages sampled for schema inference.	No	Default: 100.
   aliyun.kafka.accessKeyId	String	AccessKey ID of your Alibaba Cloud account. For more information, see Create an AccessKey pair.	No	Required for ApsaraMQ for Kafka clusters. Note Only VVR 6.0.2 or later support this parameter.
   aliyun.kafka.accessKeySecret	String	AccessKey secret of your Alibaba Cloud account. For more information, see Create an AccessKey pair.	No	Required for ApsaraMQ for Kafka clusters. Note Only VVR 6.0.2 or later support this parameter.
   aliyun.kafka.instanceId	String	ApsaraMQ for Kafka instance ID. Find it on the instance details page in the ApsaraMQ for Kafka console.	No	Required for ApsaraMQ for Kafka clusters. Note Only VVR 6.0.2 or later support this parameter.
   aliyun.kafka.endpoint	String	ApsaraMQ for Kafka API endpoint. For more information, see Endpoints.	No	Required for ApsaraMQ for Kafka clusters. Note Only VVR 6.0.2 or later support this parameter.
   aliyun.kafka.regionId	String	Region ID of the Kafka instance. For more information, see Endpoints.	No	Required for ApsaraMQ for Kafka clusters. Note Only VVR 6.0.2 or later support this parameter.

2. Select the CREATE CATALOG statement, then click Run.

   

3. Verify the catalog appears in the Catalogs area on the left.

View a Kafka JSON catalog

1. In the SQL editor on Scripts, enter the following statement.

   DESCRIBE `${catalog_name}`.`${db_name}`.`${topic_name}`;

   Parameter	Description
   ${catalog_name}	The catalog name.
   ${db_name}	The cluster name.
   ${topic_name}	The topic name.

2. Run the statement to view the schema.

   

Use a Kafka JSON catalog

After you create a Kafka JSON catalog, reference its topics in Flink SQL jobs.

Use as a source table

Scenario: Extract Kafka data and write it to another system.

How to use: Query the catalog table directly:

-- Insert data from Kafka topic to target table
INSERT INTO ${other_sink_table}
SELECT order_id, product_name, quantity * price AS total_amount
FROM `${kafka_catalog}`.`${db_name}`.`${topic_name}`
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */;

Use with CTAS to sync data

Scenario: Sync entire Kafka topics without writing DDL statements.

How it works: CREATE TABLE AS (CTAS) statement creates a target table with the same schema as the source. Useful for:

• Syncing without manual schema definition

• Auto-handling schema changes

Sync a single topic:

-- Create target table with inferred schema
CREATE TABLE IF NOT EXISTS `${target_table_name}`
WITH (
  'connector' = 'hologres',
  'dbname' = 'my_database',
  'tablename' = 'orders'
)
AS TABLE `${kafka_catalog}`.`${db_name}`.`${topic_name}`
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */;

Sync multiple topics in one job:

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `target_catalog`.`target_db`.`orders`
AS TABLE `kafka_catalog`.`kafka`.`orders`
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */;

CREATE TABLE IF NOT EXISTS `target_catalog`.`target_db`.`products`
AS TABLE `kafka_catalog`.`kafka`.`products`
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */;

CREATE TABLE IF NOT EXISTS `target_catalog`.`target_db`.`customers`
AS TABLE `kafka_catalog`.`kafka`.`customers`
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */;

END;

Requirements for syncing multiple topics:

Example: The following image shows which configurations meet the requirements:

Delete a Kafka JSON catalog

1. In the SQL editor on Scripts, enter the following statement.

   DROP CATALOG ${catalog_name};

   Replace ${catalog_name} with your catalog name.

2. Select the DROP CATALOG statement, right-click, and select Run.

3. Verify the catalog no longer appears in the Catalogs area.

Reference: Schema inference details

This section provides technical details about how Kafka JSON catalogs infer table schemas.

Schema inference process

When you query a Kafka topic, Flink samples messages (up to max.fetch.records, default 100) and merges their schemas.

CREATE TEMPORARY TABLE tempTable (
    value_name STRING,
    value_ts STRING
) LIKE `kafkaJsonCatalog`.`kafka`.`testTopic`;

• Default table parameters

  Parameter	Description	Remarks
  connector	Connector type.	Value: kafka or upsert-kafka.
  topic	Topic name.	Same as table name.
  properties.bootstrap.servers	Kafka broker addresses.	Same as the catalog's properties.bootstrap.servers.
  value.format	Serialization format for message values.	Always JSON.
  value.fields-prefix	Prefix for message value fields to avoid naming conflicts.	Same as the catalog's value.fields-prefix.
  value.json.infer-schema.flatten-nested-columns.enable	Recursively expand nested columns in message values.	Same as the catalog's infer-schema.flatten-nested-columns.enable.
  value.json.infer-schema.primitive-as-string	Infer all primitive types as String for message values.	Same as the catalog's infer-schema.primitive-as-string.
  value.fields-include	Policy for handling key fields in message values.	Must be EXCEPT_KEY, meaning message values exclude key fields. You must configure this parameter if the message key is not empty. Do not configure this parameter if the message key is empty.
  key.format	The format used by the Flink Kafka connector to serialize or deserialize the Kafka message key.	Must be json or raw. Required when the message key is non-empty. If the message key is not empty but cannot be parsed, set this parameter to raw. If parsing is successful, set this parameter to json.
  key.fields-prefix	Prefix for message key fields to avoid naming conflicts.	Same as the catalog's key.fields-prefix. Required when the message key is non-empty.
  key.fields	Fields storing parsed message key data.	Automatically populated. Required when the message key is non-empty and the table is not an Upsert Kafka table.