实时计算Flink版：管理Kafka JSON Catalog

使用限制

• Kafka JSON Catalog仅支持消息格式为JSON的Topic，暂不支持其他格式。

• 仅Flink计算引擎VVR 6.0.2及以上版本支持配置Kafka JSON Catalog。

  说明

  如果您使用的是VVR 4.x版本，建议升级作业至VVR 6.0.2及以上版本后使用Kafka JSON Catalog。

• 不支持通过DDL语句修改已有的Kafka JSON Catalog。

• 仅支持查询数据表，不支持创建、修改和删除数据库和表。

  说明

  CDAS或CTAS的Kafka JSON Catalog场景下，可以自动创建Topic。

• Kafka JSON Catalog不支持读取或写入开启了SSL或SASL认证的Kafka。

• Kafka JSON Catalog提供的表可以直接作为Flink SQL作业中的源表，不支持作为结果表和Lookup维表。

• 由于云消息队列 Kafka 版暂不支持采用开源版Kafka相同的接口删除Group，创建Kafka JSON Catalog时需要指定aliyun.kafka.instanceId、aliyun.kafka.accessKeyId、aliyun.kafka.accessKeySecret、aliyun.kafka.endpoint和aliyun.kafka.regionId才能自动删除Group ID，详情请参见开源对比。

注意事项

Kafka JSON Catalog通过取样数据解析生成表的Schema，对于数据格式不一致的Topic，默认保留全部列返回最宽的Schema。如果Topic的数据格式出现变化，Kafka JSON Catalog在不同时间得到的表的Schema会出现不一致。当在作业重启前后推断的Schema不同时，可能出现作业运行问题。

例如，Flink SQL作业中引用了Kafka JSON Catalog中的一张表，运行一段时间后从Savepoint重启，可能会重新获取到与上一次运行不一样的Schema。SQL作业的执行计划会用之前的Schema生成的版本，从而导致下游比如过滤条件、字段取值不匹配。因此建议在Flink SQL 作业中使用Create Temporary Table的方式建立Kakfka表，通过固定的Schema使用。

创建Kafka JSON Catalog

1. 在数据查询文本编辑区域，输入以下配置Kafka JSON Catalog的命令。

   • 自建Kafka集群或EMR 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'
     );

   • 阿里云消息队列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>'
     );

   参数	类型	说明	是否必填	备注
   YourCatalogName	String	Kafka JSON Catalog名称。	是	请填写为自定义的英文名。 重要 参数替换为您的Catalog名称后，需要去掉尖括号（<>），否则语法检查会报错。
   type	String	Catalog类型。	是	固定值为kafka。
   properties.bootstrap.servers	String	Kafka Broker地址。	是	格式为host1:port1,host2:port2,host3:port3。 以英文逗号（,）分割。
   format	String	Kafka消息格式。	是	目前只支持配置为JSON。Flink会解析JSON格式的Kafka消息，来获取Schema。
   default-database	String	Kafka集群名称。	否	默认值为kafka。Catalog要求三层结构定位一张表，即catalog_name.db_name.table_name。此处是配置默认的db_name，由于Kafka没有Database的概念，您可以在此处使用任意字符串指代Kafka集群作为database的定义。
   key.fields-prefix	String	自定义添加到消息键（Key）解析出字段名称的前缀，来避免Kafka消息键解析后的命名冲突问题。	否	默认值为key_。例如，如果您的key字段名为a，则系统默认解析key后的字段名称为key_a。 说明 key.fields-prefix的配置值不可以是value.fields-prefix的配置值的前缀。例如value.fields-prefix配置为test1_value_，则key.fields-prefix不可以配置为test1_。
   value.fields-prefix	String	自定义添加到消息体（Value）解析出字段名称的前缀，来避免Kafka消息体解析后的命名冲突问题。	否	默认值为value_。例如，如果您的value字段名为b，则系统默认解析value后的字段名称为value_b。 说明 value.fields-prefix的配置值不可以是key.fields-prefix的配置值的前缀。例如key.fields-prefix配置为test2_value_，则value.fields-prefix不可以配置为test2_。
   timestamp-format.standard	String	解析JSON格式消息中Timestamp类型字段的格式，首先尝试通过您配置的格式去解析，解析失败后再自动尝试使用其他格式解析。	否	可配置的值有以下两种： SQL（默认值） ISO-8601
   infer-schema.flatten-nested-columns.enable	Boolean	解析JSON格式消息体（Value）时，是否递归式地展开JSON中的嵌套列。	否	参数取值如下： true：递归式展开。 对于被展开的列，Flink使用索引该值的路径作为名字。例如，对于{"nested": {"col": true}} 中的列col，它展开后的名字为nested.col。 说明 设置为true时，建议和CREATE TABLE AS（CTAS）语句配合使用，目前暂不支持其它DML语句自动展开嵌套列。 false（默认值）：将嵌套类型当作String处理。
   infer-schema.primitive-as-string	Boolean	解析JSON格式消息体（Value）时，是否推导所有基本类型为String类型。	否	参数取值如下： true：推导所有基本类型为String。 false（默认值）：按照基本规则进行推导。
   infer-schema.parse-key-error.field-name	String	解析JSON格式消息键（Key）时，如果消息键不为空，且解析失败，会添加key.fields-prefix前缀拼接此配置项的值为列名，类型为VARBINARY的字段到表Schema，表示消息键部分的数据。	否	默认值为col。如：消息体解析出的字段为value_name，消息键不为空但解析失败，则默认返回的Schema包含两个字段：key_col，value_name。
   infer-schema.compacted-topic-as-upsert-table	Boolean	当Kafka topic的日志清理策略为compact且消息键（Key）不为空时，是否作为Upsert Kafka表使用。	否	默认值为true。使用CTAS或CDAS语法同步数据到阿里云消息队列Kafka版时需要配置为true。 说明 仅实时计算引擎VVR 6.0.2及以上版本支持该参数。
   max.fetch.records	Int	解析JSON格式消息时，最多尝试消费的消息数量。	否	默认值为100。
   aliyun.kafka.accessKeyId	String	阿里云账号AccessKey ID，详情请参见创建AccessKey。	否	使用CTAS或CDAS语法同步数据到阿里云消息队列Kafka版时需要配置。 说明 仅实时计算引擎VVR 6.0.2及以上版本支持该参数。
   aliyun.kafka.accessKeySecret	String	阿里云账号AccessKey Secret，详情请参见创建AccessKey。	否	使用CTAS或CDAS语法同步数据到阿里云消息队列Kafka版时需要配置。 说明 仅实时计算引擎VVR 6.0.2及以上版本支持该参数。
   aliyun.kafka.instanceId	String	阿里云Kafka消息队列实例ID，可在消息队列Kafka实例详情界面查看。	否	使用CTAS或CDAS语法同步数据到阿里云消息队列Kafka版时需要配置。 说明 仅实时计算引擎VVR 6.0.2及以上版本支持该参数。
   aliyun.kafka.endpoint	String	阿里云Kafka API服务接入地址，详情请参见服务接入点。	否	使用CTAS或CDAS语法同步数据到阿里云消息队列Kafka版时需要配置。 说明 仅实时计算引擎VVR 6.0.2及以上版本支持该参数。
   aliyun.kafka.regionId	String	Topic所在实例的地域ID，详情请参见服务接入点。	否	使用CTAS或CDAS语法同步数据到阿里云消息队列Kafka版时需要配置。 说明 仅实时计算引擎VVR 6.0.2及以上版本支持该参数。

2. 选中创建Catalog的代码后，单击左侧代码行数上的运行。

   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. 在左侧元数据区域，查看创建的Catalog。

查看Kafka JSON Catalog

1. 在数据查询文本编辑区域，输入以下命令。

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

   参数	说明
   ${catalog_name}	Kafka JSON Catalog名称。
   ${db_name}	Kafka集群名称。
   ${topic_name}	Kafka Topic名称。

2. 选中查看Catalog的代码后，单击左侧代码行数上的运行。

   运行成功后，可以在运行结果中查看表的具体信息。执行后返回Kafka源表的字段定义，包含业务字段value_from（STRING）、value_name（STRING）、value_order（BIGINT），以及元数据虚拟字段partition（INT，主键）、offset（BIGINT，主键）、timestamp（TIMESTAMP_LTZ(3)），其中元数据字段的extras列均为METADATA VIRTUAL。

使用Kafka JSON Catalog

• 作为源表，从Kafka Topic中读取数据。

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

  说明

  如果Kafka JSON Catalog的表使用时需要指定其他WITH参数，则建议使用SQL Hints的方式来添加其他参数。例如，如上SQL使用了SQL Hints指定从最早的数据开始消费。其他参数详情请参见消息队列Kafka源表和消息队列Kafka结果表。

• 作为源表，使用CREATE TABLE AS（CTAS）语句将Kafka Topic中的数据同步至目标表中。

  • 单表同步，实时同步数据。

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

  • 在一个作业中同步多张表。

    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;

    结合Kafka JSON Catalog，您可以在同一个任务中同步多张Kafka表。但需要满足以下条件：

    • 所有Kafka表均未配置topic-pattern参数。

    • 每张表关于Kafka的配置必须完全相同，即properties.*配置的属性完全相同，包括properties.bootstrap.servers和properties.group.id。

    • 每张表的 scan.startup.mode配置必须完全相同，且只能配置为group-offsets、latest-offset或earliest-offset，不能配置为其他值。

    例如，下图中上面两张表满足条件，下面两张表违反了以上三个条件。

删除Kafka JSON Catalog

1. 在数据查询文本编辑区域，输入以下命令。

   DROP CATALOG ${catalog_name};

   其中${catalog_name}为您要删除的目标Kafka JSON Catalog名称。

2. 选中删除Catalog的命令，鼠标右键选择运行。

3. 在左侧元数据区域，查看目标Catalog是否已删除。

从Kafka JSON Catalog获取的表信息详解