Realtime Compute for Apache Flink:Kafka JSON Catalogs

Limitations

• A Kafka JSON catalog only supports topics with messages in JSON format.

• Only Flink computing engine VVR 6.0.2 or later supports configuring Kafka JSON catalogs.

  Note

  To use Kafka JSON catalogs on VVR 4.x, upgrade your jobs to VVR 6.0.2 or later.

• You cannot use DDL statements to modify an existing Kafka JSON catalog.

• You can only query data tables. You cannot create, modify, or delete databases or tables.

  Note

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

• A Kafka JSON catalog cannot read from or write to Kafka clusters with SSL or SASL authentication.

• You can use tables from a Kafka JSON catalog directly as source tables in Flink SQL jobs, but not as sink tables or lookup dimension tables.

• Because ApsaraMQ for Kafka currently does not support deleting a Group by using the same interface as open-source Kafka, you must specify aliyun.kafka.instanceId, aliyun.kafka.accessKeyId, aliyun.kafka.accessKeySecret, aliyun.kafka.endpoint, and aliyun.kafka.regionId when you create a Kafka JSON catalog to enable automatic deletion of the Group ID. For more information, see Comparison with open-source Kafka.

Usage notes

The Kafka JSON Catalog infers a table's schema by parsing sample data. When a topic contains inconsistent data formats, the catalog keeps all columns by default and returns the widest schema. Changes to the data format in the topic can alter the inferred schema, which may cause job failures on restart.

For example, when restarting a Flink SQL job that reads from a Kafka JSON Catalog table from a savepoint, the catalog might infer a new schema that differs from the previous one. The job's execution plan is based on the old schema, which can cause runtime errors such as mismatched filter conditions or incorrect field access. To prevent these issues, use the CREATE TEMPORARY TABLE statement in your Flink SQL job to define a Kafka table with a fixed schema.

Create a Kafka JSON catalog

1. On the Scripts page, enter the statement to create a Kafka JSON catalog in the SQL editor.

   • For a self-managed Kafka cluster or an EMR on ECS Kafka cluster

     CREATE CATALOG <YourCatalogName> WITH(
      'type'='kafka',
      'properties.bootstrap.servers'='<brokers>',
      'format'='json',
      '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'
     );

   • For ApsaraMQ for Kafka

     CREATE CATALOG <YourCatalogName> WITH(
      'type'='kafka',
      'properties.bootstrap.servers'='<brokers>',
      'format'='json',
      '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>',
      'aliyun.kafka.accessKeySecret'='<aliyunAccessKeySecret>',
      'aliyun.kafka.instanceId'='<aliyunKafkaInstanceId>',
      'aliyun.kafka.endpoint'='<aliyunKafkaEndpoint>',
      'aliyun.kafka.regionId'='<aliyunKafkaRegionId>'
     );

   Parameter	Type	Description	Required	Remarks
   YourCatalogName	String	The name of the Kafka JSON catalog.	Yes	Specify a custom name. Important Replace the placeholder with your catalog name and remove the angle brackets (<>) to prevent a syntax error.
   type	String	The type of the catalog.	Yes	The value must be 'kafka'.
   properties.bootstrap.servers	String	The Kafka broker addresses.	Yes	Specify one or more addresses in the host1:port1,host2:port2,host3:port3 format. Separate multiple addresses with commas (,).
   format	String	The Kafka message format.	Yes	Currently, only 'json' is supported. Flink parses JSON-formatted messages to infer the schema.
   default-database	String	The default database name for the catalog.	No	Catalogs use a three-part identifier to locate tables: catalog_name.db_name.table_name. This parameter sets the db_name. Since Kafka does not have databases, you can use any string to represent the cluster.
   key.fields-prefix	String	A custom prefix added to the names of fields parsed from the message key to prevent naming conflicts.	No	The default value is 'key_'. For example, if a field in the key is named 'a', its name in the table schema becomes 'key_a'. Note The value of key.fields-prefix cannot be a prefix of the value of value.fields-prefix. For example, if value.fields-prefix is set to 'test1_value_', you cannot set key.fields-prefix to 'test1_'.
   value.fields-prefix	String	A custom prefix added to the names of fields parsed from the message value to prevent naming conflicts.	No	The default value is 'value_'. For example, if a field in the value is named 'b', its name in the table schema becomes 'value_b'. Note The value of value.fields-prefix cannot be a prefix of the value of key.fields-prefix. For example, if key.fields-prefix is set to 'test2_value_', you cannot set value.fields-prefix to 'test2_'.
   timestamp-format.standard	String	The format for parsing timestamp fields in JSON messages. Flink first attempts to parse timestamps using this format. If parsing fails, it automatically tries other standard formats.	No	Valid values: SQL (default) ISO-8601
   infer-schema.flatten-nested-columns.enable	Boolean	Specifies whether to recursively expand nested columns in the JSON message value.	No	Valid values: true: Recursively expands nested columns. For expanded columns, Flink uses the access path as the column name. For example, for a column 'col' in {"nested": {"col": true}}, the expanded column name is 'nested.col'. Note If this parameter is set to true, we recommend that you use it with the CREATE TABLE AS (CTAS) statement. Other DML statements do not support automatic expansion of nested columns. false (default): Treats nested types as STRING.
   infer-schema.primitive-as-string	Boolean	Specifies whether to infer all primitive types in the JSON message value as STRING.	No	Valid values: true: Infers all primitive types as STRING. false (default): Infers types based on standard rules.
   infer-schema.parse-key-error.field-name	String	If a non-empty message key cannot be parsed as JSON, Flink adds a VARBINARY field to the schema to hold the raw key data. The name of this new field is a combination of the key.fields-prefix and this parameter's value.	No	The default value is 'col'. For example, if the message value parses into a field named 'value_name' and the message key is not empty but fails to be parsed, the resulting schema contains two fields: 'key_col' and 'value_name'.
   infer-schema.compacted-topic-as-upsert-table	Boolean	Specifies whether to treat the topic as an Upsert Kafka table. This applies only when the topic's cleanup policy is 'compact' and the message key is not empty.	No	The default value is true. This parameter must be set to true when you use CTAS or CDAS statements to synchronize data to an ApsaraMQ for Kafka instance. Note This parameter is supported only in VVR 6.0.2 and later.
   max.fetch.records	Int	The maximum number of messages to sample for schema inference.	No	The default value is 100.
   aliyun.kafka.accessKeyId	String	The AccessKey ID of your Alibaba Cloud account. For more information, see Create an AccessKey pair.	No	This parameter is required when you use CTAS or CDAS statements to synchronize data to an ApsaraMQ for Kafka instance. Note This parameter is supported only in VVR 6.0.2 and later.
   aliyun.kafka.accessKeySecret	String	The AccessKey Secret of your Alibaba Cloud account. For more information, see Create an AccessKey pair.	No	This parameter is required when you use CTAS or CDAS statements to synchronize data to an ApsaraMQ for Kafka instance. Note This parameter is supported only in VVR 6.0.2 and later.
   aliyun.kafka.instanceId	String	The instance ID of the ApsaraMQ for Kafka instance. You can view the instance ID on the instance details page in the ApsaraMQ for Kafka console.	No	This parameter is required when you use CTAS or CDAS statements to synchronize data to an ApsaraMQ for Kafka instance. Note This parameter is supported only in VVR 6.0.2 and later.
   aliyun.kafka.endpoint	String	The API endpoint for ApsaraMQ for Kafka. For more information, see Endpoints.	No	This parameter is required when you use CTAS or CDAS statements to synchronize data to an ApsaraMQ for Kafka instance. Note This parameter is supported only in VVR 6.0.2 and later.
   aliyun.kafka.regionId	String	The Region ID of the instance where the topic resides. For more information, see Endpoints.	No	This parameter is required when you use CTAS or CDAS statements to synchronize data to an ApsaraMQ for Kafka instance. Note This parameter is supported only in VVR 6.0.2 and later.

2. Select the CREATE CATALOG statement, and then click Run in the left-side gutter.

   CREATE CATALOG KafkaCatalog WITH(
   'type'='kafka',
   'properties.bootstrap.servers'='<brokers>',
   'format'='json',
   '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>',
   'aliyun.kafka.accessKeySecret'='<aliyunAccessKeySecret>',
   'aliyun.kafka.instanceId'='<aliyunKafkaInstanceId>',
   'aliyun.kafka.endpoint'='<aliyunKafkaEndpoint>',
   'aliyun.kafka.regionId'='<aliyunKafkaRegionId>'
   );

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

Kafka JSON catalog

1. Enter the following command in the Scripts editor.

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

   Parameter	Description
   ${catalog_name}	The name of the Kafka JSON catalog.
   ${db_name}	The name of the Kafka cluster.
   ${topic_name}	The name of the Kafka topic.

2. Select the DESCRIBE statement, and then click Run in the left-side gutter.

   After the statement runs successfully, the results pane displays the field definitions for the Kafka source table. The output includes the business fields value_from (STRING), value_name (STRING), and value_order (BIGINT), and the metadata virtual fields partition (INT, primary key), offset (BIGINT, primary key), and timestamp (TIMESTAMP_LTZ(3)). For these metadata virtual fields, the extras column shows METADATA VIRTUAL.

Use a Kafka JSON catalog

• As a source table to read data from a Kafka topic.

  INSERT INTO ${other_sink_table}
  SELECT...
  FROM `${kafka_catalog}`.`${db_name}`.`${topic_name}`/*+OPTIONS('scan.startup.mode'='earliest-offset')*/;

  Note

  To specify additional WITH options for a table in the Kafka JSON catalog, use SQL hints. For example, the preceding SQL statement uses a hint to start consuming data from the earliest offset. For more information about other parameters, see ApsaraMQ for Kafka source table parameters and ApsaraMQ for Kafka sink table parameters.

• As a source table with a CREATE TABLE AS (CTAS) statement to synchronize data from a Kafka topic to a destination table.

  • Synchronize a single table in real time.

    CREATE TABLE IF NOT EXISTS `${target_table_name}`
    WITH(...)
    AS TABLE `${kafka_catalog}`.`${db_name}`.`${topic_name}`
    /*+OPTIONS('scan.startup.mode'='earliest-offset')*/;

  • Synchronize multiple tables in a single job.

    BEGIN STATEMENT SET;
    CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table0`
    AS TABLE `kafka-catalog`.`kafka`.`topic0`
    /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
    CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table1`
    AS TABLE `kafka-catalog`.`kafka`.`topic1`
    /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
    CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table2`
    AS TABLE `kafka-catalog`.`kafka`.`topic2`
    /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
    END;

    With a Kafka JSON catalog, you can synchronize multiple Kafka tables in the same job. The following conditions apply:

    • None of the Kafka tables are configured with the topic-pattern parameter.

    • The Kafka configuration for each table must be identical, including all properties.* settings such as properties.bootstrap.servers and properties.group.id.

    • The scan.startup.mode setting for each table must be identical and can only be set to group-offsets, latest-offset, or earliest-offset.

    For example, in the following figure, the top two tables meet the conditions, while the bottom two tables violate them.

Delete a Kafka JSON catalog

1. Enter the following command in the SQL editor on the Scripts page.

   DROP CATALOG ${catalog_name};

   Replace ${catalog_name} with the name of your Kafka JSON catalog.

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

3. Verify that the catalog is no longer listed in the Catalogs area on the left.

Kafka JSON Catalog tables