Paimonテーブルのデータの書き込みと消費 - Realtime Compute for Apache Flink

前提条件

まず、Paimon カタログと Paimon テーブルを作成する必要があります。詳細については、「Paimon カタログの管理」をご参照ください。

制限事項

Paimon テーブルは、Ververica Runtime (VVR) 8.0.5 以降でのみサポートされています。

Paimon テーブルへのデータの書き込み

CTAS/CDAS を使用したデータとスキーマの同期

詳細については、「Paimon カタログの管理」をご参照ください。

INSERT INTO を使用したデータの挿入または更新

INSERT INTO ステートメントを使用して、Paimon テーブルにデータを直接挿入または更新できます。

Paimon プライマリーキーテーブルは、INSERT、UPDATE_BEFORE、UPDATE_AFTER、DELETE を含むすべてのタイプのメッセージを受け入れます。データマージメカニズムに基づき、書き込み時に同じプライマリーキーを持つデータをマージします。
Paimon アペンドオンリーテーブル (非プライマリーキーテーブルとも呼ばれます) は、INSERT タイプのメッセージのみを受け入れます。

INSERT OVERWRITE を使用したデータの上書き

上書き操作は、既存のデータをクリアして新しいデータを書き込みます。INSERT OVERWRITE ステートメントを使用して、Paimon テーブル全体または特定のパーティションを上書きできます。以下に例を示します：

説明

INSERT OVERWRITE ステートメントは、バッチジョブでのみサポートされています。
デフォルトでは、INSERT OVERWRITE 操作はチェンジログデータを生成しません。ダウンストリームのストリーミングジョブは、削除およびインポートされたデータを消費できません。このタイプのデータを消費する必要がある場合は、「INSERT OVERWRITE の結果のストリーミングと消費」をご参照ください。

非パーティションテーブル my_table 全体を上書きします。
```
INSERT OVERWRITE my_table SELECT ...;
```
my_table テーブルの dt=20240108,hh=06 パーティションを上書きします。
```
INSERT OVERWRITE my_table PARTITION (`dt` = '20240108', `hh` = '06') SELECT ...;
```
my_table テーブルのパーティションを動的に上書きします。SELECT ステートメントの結果に含まれるパーティションが上書きされ、他のパーティションは変更されません。
```
INSERT OVERWRITE my_table SELECT ...;
```

パーティションテーブル my_table 全体を上書きします。

INSERT OVERWRITE my_table /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...;

DELETE を使用したデータの削除

DELETE ステートメントを使用して、Paimon プライマリーキーテーブルからデータを削除できます。DELETE ステートメントは [Data Exploration] でのみ実行できます。

-- my_table テーブルから currency = 'UNKNOWN' のデータをすべて削除します。
DELETE FROM my_table WHERE currency = 'UNKNOWN';

DELETE メッセージのフィルタリング

Paimon プライマリーキーテーブルを使用する場合、DELETE メッセージはデフォルトで対応するプライマリーキーを持つデータを削除します。Paimon テーブルがこれらのメッセージを処理しないようにするには、SQL ヒントを使用して次のパラメーターを true に設定します。これにより、DELETE メッセージをフィルタリングします。

パラメーター	説明	型	デフォルト
ignore-delete	DELETE メッセージをフィルタリングするかどうかを指定します。	Boolean	false

シンクの並列度の調整

SQL ヒントを使用して次のパラメーターを設定して、シンクオペレーターの並列度を調整できます：

パラメーター	説明	型	デフォルト
sink.parallelism	Paimon シンクオペレーターの並列度を設定します。	Integer	None

たとえば、次の SQL ステートメントは、Paimon シンクオペレーターの並列度を 10 に設定します：

INSERT INTO t /*+ OPTIONS('sink.parallelism' = '10') */ SELECT * FROM s;

Paimon テーブルからのデータの消費

ストリーミングジョブ

説明

ストリーミングジョブで Paimon プライマリーキーテーブルのデータを消費する場合、チェンジログプロデューサーを設定する必要があります。

デフォルトでは、ストリーミングジョブ内の Paimon ソースオペレーターは、ジョブの開始時にまずテーブルの全量データを生成し、それ以降は増分データを生成します。

指定されたオフセットからのデータの消費

次のいずれかの方法で、指定されたオフセットからデータを消費できます：

ジョブの起動時に Paimon テーブルの全量データを消費する必要がなく、後続の増分データのみを消費する必要がある場合は、SQL ヒントで 'scan.mode' = 'latest' を設定します。
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest') */;
```
全量データを消費せず、特定の時点からの増分データのみを消費したい場合は、SQL ヒントを使用して scan.timestamp-millis パラメーターを設定できます。このパラメーターの値は、UNIX エポック (1970-01-01 00:00:00 UTC) から指定された時点までのミリ秒数を表します。
```
SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;
```
特定の時刻以降に書き込まれたデータを消費し、その後継続的に増分データを消費するには、次のいずれかの方法を使用します：

説明
この消費方法は、指定された時刻以降に変更されたデータファイルを読み取ります。コンパクションのため、データファイルには指定された時刻より前に書き込まれた少量のデータが含まれる場合があります。必要に応じて、SQL ジョブに WHERE 句を追加してデータをフィルタリングできます。
- SQL ヒントは設定しないでください。ジョブを開始するときに、[Specify source's start time] を選択できます。[Job Start] ダイアログボックスで [Stateless start-up] を選択し、[Specify source's start time] スイッチを有効にして、対象の時刻を設定します。
- SQL ヒントを使用して scan.file-creation-time-millis パラメーターを設定します。
```
SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') */;
```
全量データを消費せず、特定のスナップショットファイルから始まる増分データのみを消費したい場合は、SQL ヒントを使用して scan.snapshot-id パラメーターを設定できます。このパラメーターの値は、指定されたスナップショットファイルの ID です。
```
SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;
```
特定のスナップショットファイルの全量データを消費し、継続して増分データを消費したい場合は、SQL ヒントを使用して 'scan.mode' = 'from-snapshot-full' と scan.snapshot-id パラメーターを設定できます。scan.snapshot-id パラメーターの値は、指定されたスナップショットファイルの ID です。
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'from-snapshot-full', 'scan.snapshot-id' = '1') */;
```

コンシューマー ID の指定

コンシューマー ID は、Paimon テーブルの消費の進捗を保存します。これは主に次のシナリオで使用されます：

コンシューマー ID を設定すると、対応する消費の進捗が Paimon テーブルのメタデータファイルに保存されます。これにより、ジョブが後でステートレスモードで開始された場合でも、中断した時点から消費を再開できます。
コンシューマー ID を設定すると、まだ消費されていないスナップショットは有効期限が切れても削除されません。これにより、消費速度がスナップショットの有効期限切れに追いつかずにエラーが発生するのを防ぎます。

consumer-id パラメーターを設定して、ストリーミングジョブ内の Paimon ソースオペレーターにコンシューマー ID を割り当てることができます。コンシューマー ID の値は任意の文字列に設定できます。コンシューマー ID が初めて作成されるとき、その開始オフセットは「指定されたオフセットからのデータの消費」のルールに従って決定されます。その後、同じコンシューマー ID を引き続き使用することで、Paimon テーブルの消費を再開できます。

たとえば、次の SQL ステートメントは、Paimon ソースオペレーターに test-id という名前のコンシューマー ID を設定します。特定のコンシューマー ID のオフセットをリセットしたい場合は、'consumer.ignore-progress' = 'true' を設定することもできます。

SELECT * FROM t /*+ OPTIONS('consumer-id' = 'test-id') */;

説明

コンシューマー ID によってまだ消費されていないスナップショットファイルは、有効期限が切れても削除されません。古いコンシューマー ID がクリーンアップされない場合、対応するスナップショットファイルと履歴データファイルは決して削除されず、ストレージ領域を消費します。consumer.expiration-time テーブルパラメーターを設定して、指定された期間使用されていないコンシューマー ID をクリーンアップできます。たとえば、'consumer.expiration-time' = '3d' は、3 日間使用されていないコンシューマー ID がクリーンアップされることを示します。

INSERT OVERWRITE の結果のストリーミングと消費

デフォルトでは、INSERT OVERWRITE 操作はチェンジログデータを生成せず、削除およびインポートされたデータはダウンストリームのストリーミングジョブでは消費できません。このようなデータを消費する必要がある場合は、SQL ヒントを使用してストリーミング消費ジョブで 'streaming-read-overwrite' = 'true' を設定できます。

SELECT * FROM t /*+ OPTIONS('streaming-read-overwrite' = 'true') */;

バッチジョブ

デフォルトでは、バッチジョブ内の Paimon ソースオペレーターは最新のスナップショットを読み取り、Paimon テーブルの最新の状態データを出力します。

バッチタイムトラベル

SQL ヒントで scan.timestamp-millis パラメーターを設定することで、特定の時点での Paimon テーブルの状態をクエリできます。このパラメーターの値は、UNIX エポック (1970-01-01 00:00:00 UTC) から指定された時刻までのミリ秒数を表します。

SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

SQL ヒントを使用して scan.snapshot-id パラメーターを指定されたスナップショットの ID に設定することで、スナップショットが作成された時点での Paimon テーブルの状態をクエリできます。

SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;

スナップショット間の変更のクエリ

2 つのスナップショット間で Paimon テーブルのデータ変更をクエリしたい場合は、SQL ヒントを使用して incremental-between パラメーターを設定できます。たとえば、スナップショット 20 とスナップショット 12 の間で変更されたすべてのデータを表示するには、SQL ステートメントは次のようになります：

SELECT * FROM t /*+ OPTIONS('incremental-between' = '12,20') */;

説明

バッチジョブは DELETE メッセージの消費をサポートしていないため、これらのメッセージはデフォルトで破棄されます。バッチジョブで DELETE メッセージを消費したい場合は、監査ログシステムテーブルをクエリしてください。例： SELECT * FROM `t$audit_log` /*+ OPTIONS('incremental-between' = '12,20') */;。

ソースの並列度の調整

デフォルトでは、Paimon はパーティション数やバケット数などの情報に基づいてソースオペレーターの並列度を自動的に推論します。SQL ヒントを使用して次のパラメーターを設定して、並列度を調整できます：

パラメーター	型	デフォルト	説明
scan.parallelism	Integer	None	Paimon ソースオペレーターの並列度を設定します。
scan.infer-parallelism	Boolean	true	Paimon ソースオペレーターの並列度を自動的に推論するかどうかを指定します。
scan.infer-parallelism.max	Integer	1024	自動的に推論される Paimon ソースオペレーターの並列度の上限。

次の SQL ステートメントは、Paimon ソースオペレーターの並列度を 10 に設定する方法の例です：

SELECT * FROM t /*+ OPTIONS('scan.parallelism' = '10') */;

Paimon テーブルのディメンションテーブルとしての使用

Paimon テーブルはディメンションテーブルとしても使用できます。ディメンションテーブル JOIN の構文の詳細については、「ディメンションテーブル JOIN ステートメント」をご参照ください。

VARIANT 型の書き込みと消費

VVR 11.1 以降では、Paimon テーブルは VARIANT 半構造化データ型をサポートしています。この型を使用すると、PARSE_JSON または TRY_PARSE_JSON を使用して VARCHAR JSON 文字列を VARIANT 型に変換できます。VARIANT 型を直接書き込んで消費することで、JSON のクエリおよび処理パフォーマンスが大幅に向上します。

以下にコード例を示します：

CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  k BIGINT,
  info VARIANT
);
INSERT INTO `my-catalog`.`my_db`.`my_tbl` 
SELECT k, PARSE_JSON(jsonStr) FROM T;

Realtime Compute for Apache Flink:Paimon テーブルへのデータの書き込みと消費