使用限制

• 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

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的代碼后，單擊左側代碼行數上的運行。

   

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 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獲取的表信息詳解