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

制限事項

• Kafka JSON カタログは、JSON 形式のトピックのみをサポートします。

• Ververica Runtime（VVR）6.0.2 以降を使用する Realtime Compute for Apache Flink のみ Kafka JSON カタログをサポートします。

  説明

  デプロイメントで VVR 4.X を使用している場合は、Kafka JSON カタログを使用する前に、デプロイメントの VVR バージョンを VVR 6.0.2 以降にアップグレードすることをお勧めします。

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

• Kafka JSON カタログを使用してデータテーブルをクエリすることのみ可能です。Kafka JSON カタログを使用してデータベースとテーブルを作成、変更、または削除することはできません。

  説明

  Kafka JSON カタログを CREATE DATABASE AS 文または CREATE TABLE AS 文と一緒に使用すると、トピックが自動的に作成されます。

• Kafka JSON カタログを使用して、SSL ベースの認証または簡易認証およびセキュリティ層（SASL）認証が有効になっている Kafka クラスタからデータを読み書きすることはできません。

• Kafka JSON カタログのテーブルは、Flink SQL デプロイメントのソーステーブルとして使用できますが、結果テーブルまたはディメンションテーブルとして使用されるルックアップテーブルとして使用することはできません。

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

注意事項

Kafka JSON カタログを作成する

1. スクリプトページの スクリプト タブのコードエディタで、次の文を入力して Kafka JSON カタログを作成します。

   • 自己管理 Kafka クラスタまたは E-MapReduce（EMR）Kafka クラスタ の Kafka JSON カタログを作成するために使用する文

     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 インスタンスの Kafka JSON カタログを作成するために使用する文

     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 ブローカーの IP アドレスまたはエンドポイント。	はい	形式: host1:port1,host2:port2,host3:port3。 複数の host:port ペアをカンマ（,）で区切ります。
   format	STRING	Kafka メッセージの形式。	はい	JSON 形式のみがサポートされています。Realtime Compute for Apache Flink は、JSON 形式の Kafka メッセージを解析してスキーマを取得します。
   default-database	STRING	Kafka クラスタの名前。	いいえ	デフォルト値: kafka。カタログは catalog_name.db_name.table_name に基づいてテーブルを定義します。db_name のデフォルト値は catalog_name.db_name.table_name で使用されます。Kafka はデータベースを提供していません。文字列を使用して、Kafka クラスタの db_name の値を変更できます。
   key.fields-prefix	STRING	Kafka メッセージのキーフィールドから解析されたフィールドに追加されるプレフィックス。このパラメータを構成すると、Kafka メッセージのキーフィールドが解析された後に名前の競合を防ぐことができます。	いいえ	デフォルト値: key_。キーフィールドの名前が a の場合、Kafka メッセージのキーフィールドが解析された後に取得されるキーの名前は key_a です。 説明 key.fields-prefix パラメータの値は、value.fields-prefix パラメータの値のプレフィックスと同じにすることはできません。たとえば、value.fields-prefix パラメータが test1_value_ に設定されている場合、key.fields-prefix パラメータを test1_ に設定することはできません。
   value.fields-prefix	STRING	Kafka メッセージの値フィールドから解析されたフィールドに追加されるプレフィックス。このパラメータを構成すると、Kafka メッセージの値フィールドが解析された後に名前の競合を防ぐことができます。	いいえ	デフォルト値: value_。値フィールドの名前が b の場合、Kafka メッセージの値フィールドが解析された後に取得される値の名前は value_b です。 説明 value.fields-prefix パラメータの値は、key.fields-prefix パラメータの値のプレフィックスと同じにすることはできません。たとえば、key.fields-prefix パラメータが test2_value_ に設定されている場合、value.fields-prefix パラメータを test2_ に設定することはできません。
   timestamp-format.standard	STRING	JSON 形式の Kafka メッセージの TIMESTAMP 型のフィールドの形式。Realtime Compute for Apache Flink は、構成した形式でフィールドを解析します。Realtime Compute for Apache Flink が構成した形式でフィールドを解析できない場合、Realtime Compute for Apache Flink は別の形式でフィールドを解析しようとします。	いいえ	有効な値: SQL（デフォルト値） ISO-8601
   infer-schema.flatten-nested-columns.enable	BOOLEAN	JSON 形式の Kafka メッセージの値フィールドが解析されるときに、JSON テキストのネストされた列を再帰的に展開するかどうかを指定します。	いいえ	有効な値: true: ネストされた列が再帰的に展開されます。 Realtime Compute for Apache Flink は、展開された列の値をインデックス付けするパスを列の名前として使用します。たとえば、{"nested": {"col": true}} の列 col は、展開された後に nested.col という名前になります。 説明 このパラメータを true に設定する場合は、このパラメータを CREATE TABLE AS 文と一緒に使用することをお勧めします。他の DML 文は、ネストされた列を自動的に展開するために使用できません。 false: ネストされた型は STRING 型として解析されます。これはデフォルト値です。
   infer-schema.primitive-as-string	BOOLEAN	JSON 形式の Kafka メッセージの値フィールドが解析されるときに、すべての基本型を STRING 型として推測するかどうかを指定します。	いいえ	有効な値: true: すべての基本型は STRING 型として推測されます。 false: データ型はデータ型マッピングに基づいて推測されます。これはデフォルト値です。
   infer-schema.parse-key-error.field-name	STRING	キーフィールドのデータ。JSON 形式の Kafka メッセージのキーフィールドが解析されるときに、キーフィールドが指定されているが解析に失敗した場合、名前が key.fields-prefix プレフィックスとこのパラメータの値である列が、トピックに一致するテーブルのスキーマに追加されます。この列は VARBINARY 型であり、キーフィールドのデータを示します。	いいえ	デフォルト値: col。たとえば、JSON 形式の Kafka メッセージの値フィールドが value_name として解析され、キーフィールドが指定されているが解析に失敗した場合、返されるトピックに一致するテーブルのスキーマには、key_col と value_name の 2 つのフィールドが含まれます。
   infer-schema.compacted-topic-as-upsert-table	BOOLEAN	Kafka トピックのログクリーンアップポリシーが compact で、キーフィールドが指定されているときに、テーブルを Upsert Kafka テーブルとして使用するかどうかを指定します。	いいえ	デフォルト値: true。CREATE TABLE AS 文または CREATE DATABASE AS 文を実行して ApsaraMQ for Kafka にデータを同期するときは、このパラメータを true に設定する必要があります。 説明 VVR 6.0.2 以降を使用する Realtime Compute for Apache Flink のみ、このパラメータをサポートします。
   max.fetch.records	INT	メッセージが解析されるときにシステムが消費を試みる JSON 形式のメッセージの最大数。	いいえ	デフォルト値: 100。
   aliyun.kafka.accessKeyId	STRING	Alibaba Cloud アカウントの AccessKey ID。AccessKey ID の取得方法については、「AccessKey ペアを作成する」をご参照ください。	いいえ	CREATE TABLE AS 文または CREATE DATABASE AS 文を実行して ApsaraMQ for Kafka にデータを同期するときは、このパラメータを構成する必要があります。 説明 VVR 6.0.2 以降を使用する Realtime Compute for Apache Flink のみ、このパラメータをサポートします。
   aliyun.kafka.accessKeySecret	STRING	Alibaba Cloud アカウントの AccessKey シークレット。AccessKey シークレットの取得方法については、「AccessKey ペアを作成する」をご参照ください。	いいえ	CREATE TABLE AS 文または CREATE DATABASE AS 文を実行して ApsaraMQ for Kafka にデータを同期するときは、このパラメータを構成する必要があります。 説明 VVR 6.0.2 以降を使用する Realtime Compute for Apache Flink のみ、このパラメータをサポートします。
   aliyun.kafka.instanceId	STRING	ApsaraMQ for Kafka インスタンスの ID。ApsaraMQ for Kafka コンソールの [インスタンスの詳細] ページで ID を表示できます。	いいえ	CREATE TABLE AS 文または CREATE DATABASE AS 文を実行して ApsaraMQ for Kafka にデータを同期するときは、このパラメータを構成する必要があります。 説明 VVR 6.0.2 以降を使用する Realtime Compute for Apache Flink のみ、このパラメータをサポートします。
   aliyun.kafka.endpoint	STRING	ApsaraMQ for Kafka のエンドポイント。詳細については、「エンドポイント」をご参照ください。	いいえ	CREATE TABLE AS 文または CREATE DATABASE AS 文を実行して ApsaraMQ for Kafka にデータを同期するときは、このパラメータを構成する必要があります。 説明 VVR 6.0.2 以降を使用する Realtime Compute for Apache Flink のみ、このパラメータをサポートします。
   aliyun.kafka.regionId	STRING	トピックが属する ApsaraMQ for Kafka インスタンスのリージョン ID。詳細については、「エンドポイント」をご参照ください。	いいえ	CREATE TABLE AS 文または CREATE DATABASE AS 文を実行して ApsaraMQ for Kafka にデータを同期するときは、このパラメータを構成する必要があります。 説明 VVR 6.0.2 以降を使用する Realtime Compute for Apache Flink のみ、このパラメータをサポートします。

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 JSON カタログのテーブルがソーステーブルとして使用されている場合、テーブルに一致する Kafka トピックからデータを読み取ることができます。

  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 結果テーブルを作成する」をご参照ください。

• Kafka JSON カタログのテーブルが Apache Kafka ソーステーブルのメッセージキューとして使用されている場合、CREATE TABLE AS 文を使用して、テーブルに一致する Kafka トピックからデスティネーションテーブルにデータを同期できます。

  • 単一のトピックからリアルタイムでデータを同期します。

    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;

    CREATE TABLE AS 文を Kafka JSON カタログと一緒に使用して、デプロイメントの複数の Kafka トピックからデータを同期できます。デプロイメントの複数の Kafka トピックからデータを同期するには、次の条件が満たされていることを確認してください。

    • topic-pattern は、トピックに一致するすべてのテーブルで構成されていません。

    • 各テーブルの Kafka パラメータの値は同じです。プレフィックスが properties. であるパラメータの値は同じです。パラメータには、properties.bootstrap.servers と properties.group.id が含まれます。

    • scan.startup.mode パラメータの値は、すべてのテーブルで同じです。scan.startup.mode パラメータは、group-offsets、latest-offset、または earliest-offset のみに設定できます。

    次の図は例を示しています。次の図では、上の 2 つのテーブルは上記の条件を満たしており、下の 2 つのテーブルは条件を満たしていません。

Kafka JSON カタログを削除する

1. スクリプトページの スクリプト タブのコードエディタで、次の文を入力します。

   DROP CATALOG ${catalog_name};

   catalog_name は、削除する Kafka JSON カタログの名前を指定します。

2. カタログを削除するために使用する文を右クリックし、ショートカットメニューから [実行] を選択します。

3. カタログリストページの左側の [カタログ] ペインで、カタログが削除されているかどうかを確認します。

スキーマ推論の説明