Realtime Compute for Apache Flink:Kafka JSON カタログの管理

制限事項

• Kafka JSON カタログは、JSON 形式のメッセージを含む Topic のみをサポートします。

• VVR 6.0.2 以降を使用するコンピュートエンジンのみが Kafka JSON カタログをサポートします。

  説明

  VVR 4.x を使用している場合、Kafka JSON カタログを使用するには、ジョブを VVR 6.0.2 以降にアップグレードする必要があります。

• DDL 文を使用して既存の Kafka JSON カタログを変更することはできません。

• データテーブルのクエリのみが可能です。データベースやテーブルの作成、変更、削除はできません。

  説明

  Kafka JSON カタログを使用する CREATE DATABASE AS (CDAS) または CREATE TABLE AS (CTAS) のシナリオでは、Topic を自動的に作成できます。

• Kafka JSON カタログは、SSL または SASL 認証が有効になっている Kafka クラスターからの読み取りや書き込みはできません。

• Kafka JSON カタログによって提供されるテーブルは、Flink SQL ジョブのソーステーブルとして直接使用できます。結果テーブルやルックアップディメンションテーブルとしては使用できません。

• ApsaraMQ for Kafka では現在、オープンソースの Apache Kafka と同じ API 操作を使用してコンシューマーグループを削除することはできません。Kafka JSON カタログを作成する際には、コンシューマーグループを自動的に削除するために、aliyun.kafka.instanceId、aliyun.kafka.accessKeyId、aliyun.kafka.accessKeySecret、aliyun.kafka.endpoint、および aliyun.kafka.regionId パラメーターを設定する必要があります。詳細については、「ApsaraMQ for Kafka とオープンソース Apache Kafka の比較」をご参照ください。

注意事項

Kafka JSON カタログは、サンプルデータを解析してテーブルスキーマを生成します。データ形式に一貫性のない Topic の場合、カタログは可能な限り広いスキーマを返し、デフォルトですべての列を保持します。Topic のデータ形式が変更されると、カタログによって推測されるテーブルスキーマは、異なる時点で一貫性がなくなる可能性があります。ジョブの再起動前後で異なるスキーマが推測された場合、ジョブの実行に問題が発生する可能性があります。

たとえば、Flink SQL ジョブが Kafka JSON カタログ内のテーブルを参照しているとします。ジョブが一定期間実行された後、セーブポイントから再起動すると、前回の実行で使用されたスキーマとは異なる新しいスキーマが取得される可能性があります。しかし、ジョブの実行計画は古いスキーマで生成されたバージョンを依然として使用します。これにより、フィルター条件やフィールド値など、下流コンポーネントで不一致が発生する可能性があります。これを防ぐには、Flink SQL ジョブで CREATE TEMPORARY TABLE 文を使用して Kafka テーブルを作成し、固定スキーマを強制します。

Kafka JSON カタログを作成する

1. [データ探索] タブのテキストエディターで、Kafka JSON カタログを設定する文を入力します。

   • 自己管理型 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'
     );

   • 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>'
     );

   パラメータ	タイプ	説明	必須	備考
   YourCatalogName	String	Kafka JSON カタログの名前。	はい	任意の英語の名前を入力します。 重要 パラメーターをカタログ名に置き換えた後、山括弧 (<>) を削除してください。削除しない場合、構文チェックエラーが発生します。
   type	String	カタログのタイプ。	はい	値は kafka である必要があります。
   properties.bootstrap.servers	String	Kafka ブローカーのアドレス。	はい	フォーマット：host1:port1,host2:port2,host3:port3。 複数のアドレスはカンマ (,) で区切ります。
   format	String	Kafka メッセージのフォーマット。	はい	現在、JSON のみがサポートされています。Flink は JSON 形式の Kafka メッセージを解析してスキーマを取得します。
   default-database	String	Kafka クラスタの名前。	いいえ	デフォルト値は kafka です。カタログがテーブルを特定するには、catalog_name.db_name.table_name という 3 部構成の名前が必要です。このパラメーターはデフォルトの db_name を指定します。Kafka にはデータベースの概念がないため、任意の文字列を使用して Kafka クラスターをデータベースとして表現できます。
   key.fields-prefix	String	メッセージキーから解析されたフィールド名に追加されるカスタムプレフィックス。これにより、Kafka メッセージキーの解析後の名前の競合を回避します。	いいえ	デフォルト値は key_ です。たとえば、キーフィールド名が a の場合、システムはフィールド名を key_a として解析します。 説明 key.fields-prefix パラメーターの値は、value.fields-prefix パラメーターの値のプレフィックスにすることはできません。たとえば、value.fields-prefix を test1_value_ に設定した場合、key.fields-prefix を test1_ に設定することはできません。
   value.fields-prefix	String	メッセージ本文 (値) から解析されたフィールド名に追加されるカスタムプレフィックス。これにより、Kafka メッセージ本文の解析後の名前の競合を回避します。	いいえ	デフォルト値は value_ です。たとえば、値フィールド名が b の場合、システムはフィールド名を 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 メッセージ本文 (値) のネストされた列を再帰的に展開するかどうかを指定します。	いいえ	有効な値: true：再帰的に展開します。 展開された列について、Flink は値をインデックスするパスを名前として使用します。たとえば、{"nested": {"col": true}} の列 col の場合、展開された名前は nested.col になります。 説明 このパラメーターを true に設定した場合、CREATE TABLE AS (CTAS) 文と組み合わせて使用してください。他の DML 文はネストされた列の自動展開をサポートしていません。 false (デフォルト)：ネストされた型を String として扱います。
   infer-schema.primitive-as-string	Boolean	JSON メッセージ本文 (値) を解析する際に、すべての基本データ型を String 型として推測するかどうかを指定します。	いいえ	有効な値: true：すべての基本データ型を String として推測します。 false (デフォルト)：基本ルールに基づいて型を推測します。
   infer-schema.parse-key-error.field-name	String	JSON メッセージキーを解析する際、キーが空でなく解析に失敗した場合、VARBINARY 型のフィールドがテーブルスキーマに追加され、メッセージキーデータを表します。列名は key.fields-prefix プレフィックスとこのパラメーターの値の連結になります。	いいえ	デフォルト値は col です。たとえば、メッセージ本文が value_name という名前のフィールドに解析され、メッセージキーが空でなく解析に失敗した場合、返されるスキーマには key_col と value_name の 2 つのフィールドが含まれます。
   infer-schema.compacted-topic-as-upsert-table	Boolean	Kafka Topic のクリーンアップポリシーが compact で、メッセージキーが空でない場合に、テーブルを Upsert Kafka テーブルとして使用するかどうかを指定します。	いいえ	デフォルト値は true です。CTAS または CDAS 構文を使用して ApsaraMQ for Kafka にデータを同期する場合は、これを true に設定します。 説明 VVR 6.0.2 以降のコンピュートエンジンのみがこのパラメーターをサポートします。
   max.fetch.records	Int	JSON 形式のメッセージを解析する際に消費するメッセージの最大数。	いいえ	デフォルト値は 100 です。
   aliyun.kafka.accessKeyId	String	ご利用の Alibaba Cloud アカウントの AccessKey ID。詳細については、「AccessKey ペアの作成」をご参照ください。	いいえ	CTAS または CDAS 構文を使用して ApsaraMQ for Kafka にデータを同期する場合に、このパラメーターを設定します。 説明 VVR 6.0.2 以降のコンピュートエンジンのみがこのパラメーターをサポートします。
   aliyun.kafka.accessKeySecret	String	ご利用の Alibaba Cloud アカウントの AccessKey Secret。詳細については、「AccessKey ペアの作成」をご参照ください。	いいえ	CTAS または CDAS 構文を使用して ApsaraMQ for Kafka にデータを同期する場合に、このパラメーターを設定します。 説明 VVR 6.0.2 以降のコンピュートエンジンのみがこのパラメーターをサポートします。
   aliyun.kafka.instanceId	String	ApsaraMQ for Kafka インスタンスの ID。ApsaraMQ for Kafka コンソールのインスタンス詳細ページで ID を表示できます。	いいえ	CTAS または CDAS 構文を使用して ApsaraMQ for Kafka にデータを同期する場合に、このパラメーターを設定します。 説明 VVR 6.0.2 以降のコンピュートエンジンのみがこのパラメーターをサポートします。
   aliyun.kafka.endpoint	String	ApsaraMQ for Kafka の API エンドポイント。詳細については、「エンドポイント」をご参照ください。	いいえ	CTAS または CDAS 構文を使用して ApsaraMQ for Kafka にデータを同期する場合に、このパラメーターを設定します。 説明 VVR 6.0.2 以降のコンピュートエンジンのみがこのパラメーターをサポートします。
   aliyun.kafka.regionId	String	Topic が配置されているインスタンスのリージョン ID。詳細については、「エンドポイント」をご参照ください。	いいえ	CTAS または CDAS 構文を使用して ApsaraMQ for Kafka にデータを同期する場合に、このパラメーターを設定します。 説明 VVR 6.0.2 以降のコンピュートエンジンのみがこのパラメーターをサポートします。

2. カタログを作成する文を選択し、左側の行番号の横にある [実行] をクリックします。

   

3. 左側の [メタデータ] エリアで、作成されたカタログを表示できます。

Kafka JSON カタログを表示する

1. [データ探索] タブのテキストエディターで、次の文を入力します。

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

   パラメータ	説明
   ${catalog_name}	Kafka JSON カタログの名前。
   ${db_name}	Kafka クラスタの名前。
   ${topic_name}	Kafka トピックの名前。

2. カタログを表示する文を選択し、左側の行番号の横にある [実行] をクリックします。

   文が正常に実行されると、実行結果でテーブルの詳細を表示できます。

Kafka JSON カタログを使用する

• カタログをソーステーブルとして使用して、Kafka Topic からデータを読み取ります。

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

  説明

  Kafka JSON カタログのテーブルを使用する際に他の WITH パラメーターを指定するには、SQL ヒントを使用して追加できます。たとえば、上記の SQL 文では、SQL ヒントを使用して最も早いデータから消費を開始するように指定しています。他のパラメーターの詳細については、「ApsaraMQ for Kafka ソーステーブル」および「ApsaraMQ for Kafka 結果テーブル」をご参照ください。

• CREATE TABLE AS (CTAS) 文を使用して、Kafka Topic をソーステーブルとして、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')*/;

  • 1 つのジョブで複数のテーブルを同期します。

    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 カタログを使用すると、同じタスクで複数の Kafka テーブルを同期できます。ただし、次の条件を満たす必要があります：

    • いずれの Kafka テーブルにも topic-pattern パラメーターが設定されていないこと。

    • 各テーブルの Kafka 構成が同一であること。つまり、properties.bootstrap.servers や properties.group.id を含むすべての properties.* 設定が同じである必要があります。

    • 各テーブルの scan.startup.mode 設定が同一であること。group-offsets、latest-offset、または earliest-offset のいずれかにのみ設定できます。

    たとえば、次の図では、上の 2 つのテーブルは条件を満たしていますが、下の 2 つのテーブルは条件に違反しています。

Kafka JSON カタログの削除

1. [データ探索] タブのテキストエディターで、次の文を入力します。

   DROP CATALOG ${catalog_name};

   ${catalog_name} を、削除したい Kafka JSON カタログの名前に置き換えます。

2. カタログを削除する文を選択し、右クリックして [実行] を選択します。

3. 左側の [メタデータ] エリアで、カタログが削除されたかどうかを確認します。

Kafka JSON カタログからのテーブル情報の詳細