Realtime Compute for Apache Flink:Manage Kafka JSON catalogs

Limits

• Kafka JSON catalogs support only JSON-formatted topics.

• Only Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 6.0.2 or later supports Kafka JSON catalogs.

  Note

  If your deployments use VVR 4.X, we recommend that you upgrade the VVR version of your deployments to VVR 6.0.2 or later before you use Kafka JSON catalogs.

• You cannot modify the existing Kafka JSON catalogs by executing DDL statements.

• You can only query data tables by using Kafka JSON catalogs. You are not allowed to create, modify, or delete databases and tables by using Kafka JSON catalogs.

  Note

  If you use Kafka JSON catalogs together with CREATE DATABASE AS statement or the CREATE TABLE AS statement, topics can be automatically created.

• You cannot use Kafka JSON catalogs to read data from or write data to Kafka clusters for which SSL-based authentication or Simple Authentication and Security Layer (SASL) authentication is enabled.

• Tables of Kafka JSON catalogs can be used as source tables in Flink SQL deployments but cannot be used as result tables or lookup tables that are used as dimension tables.

• ApsaraMQ for Kafka does not allow you to call the API operation that is the same as the API operation used by Apache Kafka to delete groups. 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 group IDs. For more information, see Comparison between ApsaraMQ for Kafka and open source Apache Kafka.

Create a Kafka JSON catalog

1. In the code editor of the Scripts tab on the Scripts page, enter the following statement to create a Kafka JSON catalog:

   • Statement used to create a Kafka JSON catalog for a self-managed Kafka cluster or an E-MapReduce (EMR) 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'
     );

   • Statement used to create a Kafka JSON catalog for an ApsaraMQ for Kafka instance

     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	Data type	Description	Required	Remarks
   YourCatalogName	STRING	The name of the Kafka JSON catalog.	Yes	Enter a custom name. Important You must remove the angle brackets (<>) when you replace the value of the parameter with the name of your catalog. Otherwise, an error is returned during the syntax check.
   type	STRING	The type of the catalog.	Yes	Set the value to kafka.
   properties.bootstrap.servers	STRING	The IP addresses or endpoints of Kafka brokers.	Yes	Format: host1:port1,host2:port2,host3:port3. Separate multiple host:port pairs with commas (,).
   format	STRING	The format of Kafka messages.	Yes	Only the JSON format is supported. Realtime Compute for Apache Flink parses JSON-formatted Kafka messages to obtain the schema.
   default-database	STRING	The name of the Kafka cluster.	No	Default value: kafka. A catalog defines a table based on catalog_name.db_name.table_name. The default value of db_name is used in catalog_name.db_name.table_name. Kafka does not provide databases. You can use a string to change the value of db_name for the Kafka cluster.
   key.fields-prefix	STRING	The prefix that is added to the field that is parsed from the key field in a Kafka message. You can configure this parameter to prevent name conflicts after the key field in the Kafka message is parsed.	No	Default value: key_. If the name of the key field is a, the name of the key that is obtained after the key field in the Kafka message is parsed is key_a. Note The value of the key.fields-prefix parameter cannot be the same as the prefix in the value of the value.fields-prefix parameter. For example, if the value.fields-prefix parameter is set to test1_value_, you cannot set the key.fields-prefix parameter to test1_.
   value.fields-prefix	STRING	The prefix that is added to the field that is parsed from a value field in a Kafka message. You can configure this parameter to prevent name conflicts after value fields in the Kafka message are parsed.	No	Default value: value_. If the name of a value field is b, the name of the value that is obtained after the value field in the Kafka message is parsed is value_b. Note The value of the value.fields-prefix parameter cannot be the same as the prefix in the value of the key.fields-prefix parameter. For example, if the key.fields-prefix parameter is set to test2_value_, you cannot set the value.fields-prefix parameter to test2_.
   timestamp-format.standard	STRING	The format of a field of the TIMESTAMP type in a JSON-formatted Kafka message. Realtime Compute for Apache Flink parses the field in the format that you configured. If Realtime Compute for Apache Flink fails to parse the field in the format that you configured, Realtime Compute for Apache Flink attempts to parse the field in another format.	No	Valid values: SQL (default value) ISO-8601
   infer-schema.flatten-nested-columns.enable	BOOLEAN	Specifies whether to recursively expand nested columns in a JSON text when a value field in the JSON-formatted Kafka message is parsed.	No	Valid values: true: Nested columns are recursively expanded. Realtime Compute for Apache Flink uses the path that indexes the value of the column that is expanded as the name of the column. For example, the column col in {"nested": {"col": true}} is named nested.col after the column is expanded. Note If you set this parameter to true, we recommend that you use this parameter together with the CREATE TABLE AS statement. Other DML statements cannot be used to automatically expand nested columns. false: Nested types are parsed as the STRING type. This is the default value.
   infer-schema.primitive-as-string	BOOLEAN	Specifies whether to infer all basic types as the STRING type when value fields in the JSON-formatted Kafka message are parsed.	No	Valid values: true: All basic types are inferred as the STRING type. false: Data types are inferred based on the data type mappings. This is the default value.
   infer-schema.parse-key-error.field-name	STRING	The data of the key field. When the key field in the JSON-formatted Kafka message is parsed, if the key field is specified but fails to be parsed, a column whose name is the key.fields-prefix prefix and the value of this parameter is added to the schema of the table that matches the topic. This column is of the VARBINARY type and indicates the data of the key field.	No	Default value: col. For example, if the value field in the JSON-formatted Kafka message is parsed as value_name and the key field is specified but fails to be parsed, the returned schema of the table that matches the topic contains two fields: key_col and value_name.
   infer-schema.compacted-topic-as-upsert-table	BOOLEAN	Specifies whether to use a table as an Upsert Kafka table when the log cleanup policy of the Kafka topic is compact and the key field is specified.	No	Default value: true. You must set this parameter to true when you execute the CREATE TABLE AS statement or the CREATE DATABASE AS statement to synchronize data to ApsaraMQ for Kafka. Note Only Realtime Compute for Apache Flink that uses VVR 6.0.2 or later supports this parameter.
   max.fetch.records	INT	The maximum number of JSON-formatted messages that the system attempts to consume when the messages are parsed.	No	Default value: 100.
   aliyun.kafka.accessKeyId	STRING	The AccessKey ID of your Alibaba Cloud account. For more information about how to obtain the AccessKey ID, see Create an AccessKey pair.	No	You must configure this parameter when you execute the CREATE TABLE AS statement or the CREATE DATABASE AS statement to synchronize data to ApsaraMQ for Kafka. Note Only Realtime Compute for Apache Flink that uses VVR 6.0.2 or later supports this parameter.
   aliyun.kafka.accessKeySecret	STRING	The AccessKey secret of your Alibaba Cloud account. For more information about how to obtain the AccessKey secret, see Create an AccessKey pair.	No	You must configure this parameter when you execute the CREATE TABLE AS statement or the CREATE DATABASE AS statement to synchronize data to ApsaraMQ for Kafka. Note Only Realtime Compute for Apache Flink that uses VVR 6.0.2 or later supports this parameter.
   aliyun.kafka.instanceId	STRING	The ID of the ApsaraMQ for Kafka instance. You can view the ID on the Instance Details page of the ApsaraMQ for Kafka console.	No	You must configure this parameter when you execute the CREATE TABLE AS statement or the CREATE DATABASE AS statement to synchronize data to ApsaraMQ for Kafka. Note Only Realtime Compute for Apache Flink that uses VVR 6.0.2 or later supports this parameter.
   aliyun.kafka.endpoint	STRING	The endpoint of ApsaraMQ for Kafka. For more information, see Endpoints.	No	You must configure this parameter when you execute the CREATE TABLE AS statement or the CREATE DATABASE AS statement to synchronize data to ApsaraMQ for Kafka. Note Only Realtime Compute for Apache Flink that uses VVR 6.0.2 or later supports this parameter.
   aliyun.kafka.regionId	STRING	The region ID of the ApsaraMQ for Kafka instance to which the topic belongs. For more information, see Endpoints.	No	You must configure this parameter when you execute the CREATE TABLE AS statement or the CREATE DATABASE AS statement to synchronize data to ApsaraMQ for Kafka. Note Only Realtime Compute for Apache Flink that uses VVR 6.0.2 or later supports this parameter.

2. Select the code that is used to create a catalog and click Run that appears on the left side of the code.

   

3. In the Catalogs pane on the left side of the Catalogs tab, view the catalog that you create.

View a Kafka JSON catalog

1. In the code editor of the Scripts tab on the Scripts page, enter the following statement:

   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 code that is used to view a catalog and click Run that appears on the left side of the code.

   After the statement is executed, you can view the information about the table that matches the topic in the result.

Use a Kafka JSON catalog

• If a table of the Kafka JSON catalog is used as a source table, you can read data from the Kafka topic that matches the table.

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

  Note

  If you need to specify other parameters in the WITH clause when you use a Kafka JSON catalog, we recommend that you use SQL hints to add other parameters. In the preceding SQL statement, SQL hints are used to specify that the consumption starts from the earliest data. For more information about other parameters, see Create an ApsaraMQ for Kafka source table and Create an ApsaraMQ for Kafka result table.

• If a table of the Kafka JSON catalog is used as a Message Queue for Apache Kafka source table, you can synchronize data from the Kafka topic that matches the table to the destination table by using the CREATE TABLE AS statement.

  • Synchronize data from a single topic 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 data from multiple topics in a deployment.

    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;

    You can use the CREATE TABLE AS statement together with Kafka JSON catalogs to synchronize data from multiple Kafka topics in a deployment. To synchronize data from multiple Kafka topics in a deployment, make sure that the following conditions are met:

    • topic-pattern is not configured for all tables that match the topics.

    • The values of Kafka parameters in each table are the same. The values of the parameters whose prefix is properties. are the same. The parameters include properties.bootstrap.servers and properties.group.id.

    • The values of the scan.startup.mode parameter are the same for all the tables. The scan.startup.mode parameter can be set only to group-offsets, latest-offset, or earliest-offset.

    The following figure shows an example. In the following figure, the upper two tables meet the preceding conditions and the lower two tables do not meet the conditions.

Drop a Kafka JSON catalog

1. In the code editor of the Scripts tab on the Scripts page, enter the following statement:

   DROP CATALOG ${catalog_name};

   catalog_name specifies the name of the Kafka JSON catalog that you want to drop.

2. Right-click the statement that is used to drop the catalog and choose Run from the shortcut menu.

3. View the Catalogs pane on the left side of the Catalog List page to check whether the catalog is dropped.

Schema inference description