MaxCompute:Realtime Compute for Apache Flink を使用した Apache Paimon 外部テーブルの作成

背景情報

Apache Paimon は、高スループットの書き込みと低レイテンシーのクエリを提供する、ストリーミングとバッチを統合したレイクストレージフォーマットです。Realtime Compute for Apache Flink や E-MapReduce の Spark、Hive、Trino などの一般的なコンピューティングエンジンは、Paimon とシームレスに統合されます。Apache Paimon を使用すると、Object Storage Service (OSS) 上にデータレイクを迅速に構築し、MaxCompute に接続してデータレイク分析を実行できます。Apache Paimon の詳細については、「Apache Paimon ドキュメント」をご参照ください。

前提条件

• MaxCompute テーブルを作成するには、操作を実行する Alibaba Cloud アカウントに CreateTable 権限が必要です。テーブル権限の詳細については、「MaxCompute 権限」をご参照ください。

• MaxCompute プロジェクトが作成されていること。詳細については、「プロジェクトの作成」をご参照ください。

• OSS が有効化されていること。バケットとファイルディレクトリが作成されていること。詳細については、「バケットの作成」をご参照ください。

  説明

  MaxCompute は特定のリージョンにのみデプロイされています。リージョン間のデータ接続の問題を防ぐため、MaxCompute プロジェクトと同じリージョンのバケットを使用することを推奨します。

• フルマネージド Flink が有効化されていること。詳細については、「ワークスペースの作成」をご参照ください。

ステップ1: Apache Paimon プラグインを MaxCompute プロジェクトにアップロードする

次のいずれかの方法を使用して、Paimon プラグインを MaxCompute プロジェクトにアップロードします。

ADD JAR <path_to_paimon_maxcompute_connector.jar>;

ステップ2: Realtime Compute for Apache Flink を使用して Apache Paimon 外部テーブルを作成する

このトピックのベストプラクティスは、Realtime Compute for Apache Flink をベースにしています。Realtime Compute for Apache Flink は、Apache Paimon ファイルのデータを OSS に書き込みます。Realtime Compute for Apache Flink コンソールで Apache Paimon カタログが作成され、Apache Paimon カタログ内に、MaxCompute が OSS 内の Apache Paimon ファイルのデータを読み取るために使用できる Apache Paimon テーブルが作成されます。次に、MaxCompute は Apache Paimon テーブルを外部テーブルとして使用して、OSS に格納されている Apache Paimon データを読み取ります。

1. Realtime Compute for Apache Flink コンソールにログオンし、Scriptを作成します。 詳細については、「クエリ スクリプト」をご参照ください。

2. スクリプトエディターに、次のカタログ定義とパラメーター値を入力します。コードを選択し、実行 をクリックします。

   CREATE CATALOG `<catalog name>` WITH (
    'type' = 'paimon',
     'metastore' = 'maxcompute',
     'warehouse' = '<warehouse>',
     'maxcompute.endpoint' = '<maxcompute.endpoint>',
     'maxcompute.project' = '<maxcompute.project>',
     'maxcompute.accessid' = '<maxcompute.accessid>',
     'maxcompute.accesskey' = '<maxcompute.accesskey>',
     'maxcompute.oss.endpoint' = '<maxcompute.oss.endpoint>',
     'fs.oss.endpoint' = '<fs.oss.endpoint>',
     'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
     'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
   );

   次の表では、コード内のパラメータについて説明します。

   パラメータ	必須	説明
   catalog name	はい	Apache Paimon カタログの名前。名前には英字のみを含めることができます。このトピックでは、カタログ名は catalogname です。
   type	はい	カタログのタイプ。値を paimon に設定します。
   metastore	はい	メタデータストレージのタイプ。値を maxcompute に設定します。
   warehouse	はい	OSS のデータウェアハウスディレクトリは、oss://<bucket>/<object> 形式です。 bucket: 作成した OSS バケットの名前。 object: データが格納されているパス。 バケット名とオブジェクト名は、OSS コンソールで確認できます。
   maxcompute.endpoint	はい	MaxCompute サービスのエンドポイント。 このパラメータは、MaxCompute プロジェクトの作成時に選択したリージョンとネットワーク接続タイプに基づいて設定する必要があります。各リージョンとネットワークタイプに対応するエンドポイントの詳細については、「エンドポイント」をご参照ください。
   maxcompute.project	はい	MaxCompute プロジェクトの名前。 スキーマ機能が有効化された MaxCompute プロジェクトはサポートされていません。
   maxcompute.accessid	はい	MaxCompute の権限を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。 AccessKey ID は、AccessKey 管理ページで取得できます。
   maxcompute.accesskey	はい	AccessKey ID に対応する AccessKey シークレット。
   maxcompute.oss.endpoint	いいえ	MaxCompute がアクセスする OSS エンドポイント。このパラメータを設定しない場合、デフォルトでは fs.oss.endpoint パラメータの値が使用されます。 重要 OSS バケットは MaxCompute プロジェクトと同じリージョンにあります。maxcompute.oss.endpoint パラメータを内部エンドポイントに設定することを推奨します。各リージョンのネットワークタイプ別の OSS エンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。
   fs.oss.endpoint	いいえ	OSS のエンドポイント。 このパラメータは、warehouse パラメータで指定された OSS バケットが Realtime Compute for Apache Flink ワークスペースと同じリージョンにない場合、または別の Alibaba Cloud アカウント内の OSS バケットが使用される場合に必要です。 説明 OSS バケットの作成時に選択したリージョンとネットワーク接続方法に基づいてエンドポイントを設定する必要があります。各リージョンとネットワークタイプに対応するエンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。
   fs.oss.accessKeyId	いいえ	OSS の読み取りおよび書き込み権限を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。 このパラメータは、warehouse パラメータで指定された OSS バケットが Realtime Compute for Apache Flink ワークスペースと同じリージョンにない場合、または別の Alibaba Cloud アカウント内の OSS バケットが使用される場合に必要です。 AccessKey ID は、AccessKey 管理ページで取得できます。
   fs.oss.accessKeySecret	いいえ	AccessKey ID に対応する AccessKey シークレット。 このパラメータは、warehouse パラメータで指定された OSS バケットが Realtime Compute for Apache Flink ワークスペースと同じリージョンにない場合、または別の Alibaba Cloud アカウント内の OSS バケットが使用される場合に必要です。

3. Apache Paimon テーブルを作成します。

   1. test_tbl という名前のテーブルを作成します。

      Script タブのスクリプトエディタで次のステートメントを実行します。結果 タブに完了メッセージが表示されるまで待ちます。この例では、test_tbl という名前のテーブルを使用します。

      CREATE TABLE `catalogname`.`default`.test_tbl (
       dt STRING,
       id BIGINT,
       data STRING,
       PRIMARY KEY (dt, id) NOT ENFORCED
      ) PARTITIONED BY (dt);

   2. テーブル test_tbl にデータを書き込みます。

      [下書き] ページで、次のステートメントを含む SQL ジョブを作成します。 次に、ジョブをデプロイして実行します。 SQL ジョブの作成方法と実行方法の詳細については、「ジョブ開発の概要」をご参照ください。

      -- チェックポイント間隔を 10 秒に設定して、データをより迅速にコミットします。
      SET 'execution.checkpointing.interval' = '10s';
      INSERT INTO `catalogname`.`default`.test_tbl VALUES ('2023-04-21', 1, 'AAA'), ('2023-04-21', 2, 'BBB'), ('2023-04-22', 1, 'CCC'), ('2023-04-22', 2, 'DDD');

      説明

      • Apache Paimon の結果テーブルは、チェックポイント処理が完了するたびにデータをコミットします。

      • 本番環境では、チェックポイント間隔とチェックポイント間の最小間隔は、レイテンシーに関するビジネス要件に基づいて異なります。ほとんどの場合、1 分から 10 分に設定されます。

      • SQL ドラフトのエンジンバージョンは vvr-8.0.5-flink-1.17 以降である必要があります。

ステップ3: MaxCompute でデータを読み取る

1. ローカルクライアント (odpscmd) を使用して接続するか、または MaxCompute SQL ステートメントを実行できる他のツールを使用して、次のコマンドを実行します。

   SET odps.sql.common.table.planner.ext.hive.bridge = true;
   SET odps.sql.hive.compatible = true;

2. 以下のコマンドを実行して、Apache Paimon 外部テーブル test_tbl からデータをクエリします:

   SELECT * FROM test_tbl WHERE dt = '2023-04-21';

   次の結果が返されます。

   +------------+------------+------------+
   | id | data | dt |
   +------------+------------+------------+
   | 1 | AAA | 2023-04-21 |
   | 2 | BBB | 2023-04-21 |
   +------------+------------+------------+