Paimon の基本的な機能 - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

前提条件

RAM ユーザーまたは RAM ロールを使用する場合、Realtime Compute for Apache Flink コンソールへのアクセスに必要な権限があることを確認してください。詳細については、「権限管理」をご参照ください。
Flink ワークスペースを作成済みであること。詳細については、「Realtime Compute for Apache Flink の有効化」をご参照ください。

Object Storage Service (OSS) を有効化し、標準ストレージクラスのバケットを作成済みであること。詳細については、「OSS コンソールの使用開始」をご参照ください。OSS は、データファイルやメタデータファイルなど、Paimon テーブルのファイルを格納するために使用されます。
Paimon テーブルは、Ververica Runtime (VVR) 8.0.5 以降でのみサポートされます。

ステップ 1：Paimon カタログの作成

スクリプトエディタに移動します。
1. Realtime Compute for Apache Flink コンソールにログインします。
2. 対象のワークスペースの [Actions] 列で、[Console] をクリックします。
3. 左側のナビゲーションバーで、[Data Development] > [Data Query] をクリックします。[Query Script] タブで、新しい [Query Script] を作成します。

エディタで、次のコードを入力して Paimon カタログを作成します。

-- my-catalog は任意のカタログ名です。
CREATE Catalog `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = '<warehouse>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

次の表に、各パラメーターの詳細を示します。

パラメーター	説明	必須	備考
type	カタログのタイプ。	はい	値は `paimon` である必要があります。
metastore	メタストアのタイプ。	はい	この例では `filesystem` を使用します。他のタイプについては、「Paimon カタログの管理」をご参照ください。
warehouse	OSS バケット内のウェアハウスディレクトリ。	はい	フォーマットは `oss://<bucket>/<object>` です。各項目の意味は次のとおりです。 bucket：作成した OSS バケットの名前。 object：データを格納するパス。バケットとオブジェクトの名前は、OSS コンソールで確認できます。
fs.oss.endpoint	OSS サービスのエンドポイント。	いいえ	`warehouse` で指定された OSS バケットが Flink ワークスペースとは異なるリージョンにある場合、または別のアリババクラウドアカウントに属している場合に指定する必要があります。詳細については、「リージョンとエンドポイント」をご参照ください。説明 Paimon テーブルを OSS-HDFS に格納する場合は、`fs.oss.endpoint`、`fs.oss.accessKeyId`、および `fs.oss.accessKeySecret` パラメーターを指定する必要があります。`fs.oss.endpoint` の値は `cn-<region>.oss-dls.aliyuncs.com` の形式です (例：`cn-hangzhou.oss-dls.aliyuncs.com`)。
fs.oss.accessKeyId	OSS への読み書き権限を持つアリババクラウドアカウントまたは RAM ユーザーの AccessKey ID。	いいえ	`warehouse` で指定された OSS バケットが Flink ワークスペースとは異なるリージョンにある場合、または別のアリババクラウドアカウントに属している場合に指定する必要があります。AccessKey ペアの取得方法については、「AccessKey ペアの作成」をご参照ください。
fs.oss.accessKeySecret	OSS への読み書き権限を持つアリババクラウドアカウントまたは RAM ユーザーの AccessKey シークレット。	いいえ

Paimon カタログを作成するコードを選択し、左側の [Run] をクリックします。

The following statement has been executed successfully! というメッセージは、カタログが作成されたことを示します。

ステップ 2：Paimon データベースとテーブルの作成

[Scripts] ページで、次のコードを入力して、my_db という名前の Paimon データベースと my_tbl という名前の Paimon テーブルを作成します。
```
CREATE DATABASE `my-catalog`.`my_db`;
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt STRING,
  id BIGINT,
  content STRING,
  PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'changelog-producer' = 'lookup'  
);
```
説明
後のストリーミング消費を有効にするため、この例では WITH 句で changelog-producer を lookup に設定しています。これにより、lookup 戦略でチェンジログが生成されます。詳細については、「チェンジログの生成」をご参照ください。
Paimon データベースと Paimon テーブルを作成するコードを選択し、左側の [Run] をクリックします。

The following statement has been executed successfully! というメッセージが返された場合、データベース my_db とテーブル my_tbl が作成されたことを示します。

ステップ 3：Paimon テーブルへのデータ書き込み

[Development] > [ETL] ページの [Drafts] タブで、空白のストリームドラフトを作成します。詳細については、「SQL ドラフトの開発」をご参照ください。次の INSERT 文を SQL エディタにコピーします。

-- Paimon は、各チェックポイントが完了した後にのみデータをコミットします。
-- データをより速くコミットするために、チェックポイントの間隔を 10 秒に短縮します。
-- 本番環境では、通常、レイテンシー要件に基づいて、チェックポイントの間隔とチェックポイント間の最小間隔を 1〜10 分に設定します。
SET 'execution.checkpointing.interval'='10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

SQL エディタの右上隅にある [Deploy] をクリックします。[Deploy draft] ダイアログボックスで、指示に従ってパラメーターを設定し、[Confirm] をクリックします。
[O&M] > [Deployments] ページで、対象のデプロイメントを見つけ、[Actions] 列の [Start] をクリックします。[Initial Mode] を選択し、[Start] をクリックします。

ジョブのステータスが [FINISHED] に変わると、データが書き込まれたことになります。

ステップ 4：Paimon テーブルからのデータストリーミング

空白のストリームドラフトを作成します。次の SQL コードを SQL エディタにコピーして、print コネクタで my_tbl テーブルのすべてのデータをログに出力します。
```
CREATE TEMPORARY TABLE Print (
  dt STRING,
  id BIGINT,
  content STRING
) WITH (
  'connector' = 'print'
);
INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;
```
SQL エディタの右上隅にある [Deploy] をクリックします。[Deploy draft] ダイアログボックスで、指示に従ってパラメーターを設定し、[Confirm] をクリックします。
[O&M] > [Deployments] ページで、対象のデプロイメントの [Actions] 列にある [Start] をクリックします。[Initial Mode] を選択し、[Start] をクリックします。
デプロイメントの詳細ページで、Flink の計算結果を確認します。
1. [O&M] > [Deployments] ページで、対象のデプロイメントの名前をクリックします。
2. [Logs] タブの [Running Logs] で、[Running Task Managers] タブに移動し、[Path, ID] 列のタスクリンクをクリックします。
3. [Stdout] をクリックして、消費された Paimon データを確認します。
標準出力 (Stdout) には、Paimon テーブルの INSERT レコードである +I[20240108, 1, apple]、+I[20240108, 2, banana]、+I[20240109, 1, cat]、および +I[20240109, 2, dog] が表示され、ストリーミング消費が成功したことを確認できます。

ステップ 5：Paimon テーブルのデータ更新

空白のストリームドラフトを作成し、次の SQL コードを SQL エディタにコピーします。

SET 'execution.checkpointing.interval' = '10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

SQL エディタの右上隅にある [Deploy] をクリックします。[Deploy draft] ダイアログボックスで、指示に従ってパラメーターを設定し、[Confirm] をクリックします。
[O&M] > [Deployments] ページで、対象のデプロイメントの [Actions] 列にある [Start] をクリックします。[Initial Mode] を選択し、[Start] をクリックします。

ジョブのステータスが [FINISHED] に変わると、データが更新されたことになります。
ステップ 4 のジョブの [Deployments] ページの [Stdout] タブで、Paimon テーブルに更新されたデータを確認します。

[Running Task Managers] タブで、対象の TaskManager をクリックし、[Stdout] サブタブを開きます。Paimon テーブルの CDC レコードである -U[20240108, 1, apple]、+U[20240108, 1, hello]、-U[20240109, 2, dog]、および +U[20240109, 2, world] が表示されます。-U は更新前のレコードを示し、+U は更新後のレコードを示します。

ステップ 6 (オプション)：ジョブのキャンセルとリソースのクリーンアップ

テスト後、ストリーミング消費ジョブをキャンセルし、リソースをクリーンアップします。

[O&M] > [Deployments] ページで、対象のデプロイメントの [Actions] 列にある [Cancel] をクリックして、ジョブを停止します。

[Scripts] タブのエディタで、次のコードを入力して、Paimon データファイルと Paimon カタログを削除します。

DROP DATABASE `my-catalog`.`my_db` CASCADE; -- Paimon データベースのすべてのデータファイルを OSS から削除します。
DROP CATALOG `my-catalog`; -- Realtime Compute for Apache Flink コンソールのメタデータから Paimon カタログを削除しますが、OSS 内のデータファイルには影響しません。

The following statement has been executed successfully! というメッセージは、Paimon データファイルとカタログが削除されたことを示します。

Realtime Compute for Apache Flink:Paimon クイックスタート：基本的な機能