Paimon コネクタ - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

このトピックでは、ストリーミングデータレイクハウス用の Paimon コネクタの使用方法を説明します。最良の結果を得るには、コネクタを Paimon カタログと組み合わせて使用することを推奨します。

背景情報

Apache Paimon は、高スループットの書き込みと低レイテンシーのクエリをサポートする、ストリーミングとバッチを統合したデータレイクストレージフォーマットです。Paimon は、Flink、Spark、Hive、Trino など、Alibaba Cloud E-MapReduce で利用できる主要なコンピューティングエンジンとシームレスに統合されます。Apache Paimon を使用すると、HDFS または OSS 上にデータレイクを迅速に構築し、これらのコンピューティングエンジンに接続してデータレイク分析を実行できます。詳細については、「Apache Paimon」をご参照ください。

カテゴリ	説明
対応タイプ	ソーステーブル、ディメンションテーブル、結果テーブル、データインジェスト用のターゲット
実行モード	ストリーミングモードとバッチモード
データフォーマット	Paimon
監視メトリクス	なし
API タイプ	データインジェスト用の SQL と YAML
結果テーブルにおける更新と削除	あり

主な機能

Apache Paimon は、以下の主な機能を提供します。

HDFS またはオブジェクトストレージ上に、軽量で低コストなデータレイクを構築できます。
ストリーミングモードとバッチモードの両方で、大規模なデータセットの読み取りと書き込みができます。
数分から数秒のデータ鮮度で、バッチおよび OLAP クエリを実行できます。
増分データの取り込みと生成が可能で、従来のオフラインデータウェアハウスと最新のストリーミングデータウェアハウスの両方に対応するストレージレイヤーとしても機能します。
事前集計データにより、ストレージコストと下流のコンピューティング負荷を削減できます。
データの履歴バージョンにアクセスできます。
データを効率的にフィルタリングできます。
スキーマ進化が可能です。

制限事項と推奨事項

Paimon コネクターを使用するには、Flink コンピューティングエンジン VVR 6.0.6 以降が必要です。

次の表に、Paimon と VVR のバージョン互換性を示します。

Apache Paimon バージョン	VVR
1.3.1	11.5, 11.6
1.3	11.4
1.2	11.2, 11.3
1.1	11.1
1.0	8.0.11
0.9	8.0.7, 8.0.8, 8.0.9, 8.0.10

同時書き込み時のストレージに関する推奨事項

複数のジョブが同一の Paimon テーブルに同時書き込みを行う場合、標準 OSS ストレージ (oss://) を使用すると、アトミックなファイル操作の制限により、コミットの競合やジョブの失敗が発生することがあります。

安定した一貫性のある書き込みを行うには、強力なアトミック性を保証するメタデータサービスまたはストレージサービスを使用してください。推奨される選択肢は Data Lake Formation (DLF) です。DLF は Paimon のメタデータとストレージを一元管理できます。代わりに、OSS-HDFS または HDFS を使用することもできます。
設定変更の反映

Paimon テーブルの設定パラメーターを変更した場合、その変更は関連するジョブを再起動して初めて有効になります。実行中のジョブでは、これらの変更は動的に読み込まれません。
削除されたパーティションの物理的な解放の遅延

DROP PARTITION 操作を実行しても、システムは基盤となる物理データファイルを直ちに削除しません。
この操作は論理削除です。Paimon は、最新のスナップショットからのみ対象パーティションのメタデータを削除します。Paimon はタイムトラベル機能をサポートしているため、履歴スナップショットからは引き続き当該パーティションのデータファイルが参照されます。物理データファイルが完全に削除されるのは、パーティションを参照しているすべての履歴スナップショットが保持期限に達し、スナップショットの有効期限切れメカニズムによってクリーンアップされてからです。

SQL

SQL ジョブで Paimon コネクタをソーステーブルまたは結果テーブルとして使用します。

構文

Paimon カタログで Paimon テーブルを作成する場合、connector パラメーターを指定する必要はありません。構文は次のとおりです：
```
CREATE TABLE ``.``.paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  ...
);
```
説明
Paimon カタログに Paimon テーブルをすでに作成している場合は、そのまま使用できます。
別のカタログで Paimon 一時テーブルを作成する場合は、'connector' と 'path' パラメーターを指定する必要があります。構文は次のとおりです：
```
CREATE TEMPORARY TABLE paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'paimon',
  'path' = '<path-to-paimon-table-files>',
  'auto-create' = 'true', -- 指定したパスに Paimon テーブルのデータファイルが存在しない場合は、自動的に作成されます。
  ...
);
```
説明
- パスの例：'path' = 'oss://<bucket>/test/order.db/orders' 。.db サフィックスは省略しないでください。Paimon は、このサフィックスに基づいてデータベースを識別します。
- 同じテーブルに書き込む複数のジョブは、同じパス設定を共有する必要があります。
- 2 つのパス設定が異なる場合、Paimon は同じテーブルとして認識しません。物理パスが同じであっても、カタログ設定に不整合があると、同時書き込み競合、失敗したコンパクション操作、データ損失につながる可能性があります。たとえば、oss://b/test と oss://b/test/ は、同じ物理的な場所を参照している場合でも、末尾のスラッシュの有無により、Paimon では異なるテーブルとして扱われます。

WITH パラメーター

パラメーター	説明	タイプ	必須	デフォルト	備考
connector	テーブルのコネクタを指定します。	String	いいえ	なし	Paimon カタログで Paimon テーブルを作成する場合、このパラメーターは必須ではありません。別のカタログで Paimon 一時テーブルを作成する場合、このパラメーターを `paimon` に設定する必要があります。
path	テーブルのストレージパス。	String	いいえ	なし	Paimon カタログで Paimon テーブルを作成する場合、このパラメーターは必須ではありません。別のカタログで Paimon 一時テーブルを作成する場合、このパラメーターは HDFS または OSS のテーブルのストレージディレクトリを指定します。
auto-create	指定されたパスにテーブルファイルが存在しない場合に自動的に作成するかどうかを指定します。	Boolean	いいえ	false	有効な値： false (デフォルト)：指定されたパスに Paimon テーブルファイルが存在しない場合、ジョブは失敗します。 true：指定されたパスが存在しない場合、Flink は Paimon テーブルファイルを自動的に作成します。
file.format	データファイルのフォーマット。	String	いいえ	parquet	有効な値： orc parquet avro lance (Realtime Compute for Apache Flink 11.6 以降でサポート)
bucket	パーティションあたりのバケット数。	Integer	いいえ	1	Paimon は `bucket-key` に基づいてデータをバケットに分散します。説明各バケットのデータサイズは 5 GB 未満にすることを推奨します。
bucket-key	バケットキーとして使用される列。	String	いいえ	なし	データをバケットに分散するために使用される列を指定します。複数の列名はコンマ (,) で区切ります。たとえば、`'bucket-key' = 'order_id,cust_id'` は、`order_id` と `cust_id` 列に基づいてデータを分散します。説明このパラメーターが指定されていない場合、Paimon はプライマリキーに基づいてデータを分散します。テーブルにプライマリキーがない場合、Paimon はすべての列の値に基づいてデータを分散します。
changelog-producer	チェンジログの生成メカニズム。	String	いいえ	none	Paimon は、任意の入力ストリームに対して完全なチェンジログを生成できます。これは、すべての `update_after` レコードに、対応する `update_before` レコードが存在することを意味します。これにより、下流での処理が簡素化されます。有効な値： none (デフォルト)：追加のチェンジログは生成されません。下流のコンシューマーは引き続きストリーミングモードで Paimon テーブルを読み取ることができますが、チェンジログは不完全です (対応する `update_before` レコードのない `update_after` レコードのみを含みます)。 input：入力ストリームをチェンジログファイルに書き込み、それが完全なチェンジログとして機能します。 full-compaction：各フルコンパクション中に完全なチェンジログを生成します。 lookup：各スナップショットがコミットされる前に完全なチェンジログを生成します。チェンジログプロデューサーの選択方法の詳細については、「Changelog Production」をご参照ください。
full-compaction.delta-commits	連続する2つのフルコンパクション間におけるコミットの最大数。	Integer	いいえ	なし	フルコンパクションをトリガーする前に許可されるスナップショットコミットの最大数を指定します。
lookup.cache-max-memory-size	Paimon ディメンションテーブルのメモリキャッシュサイズ。	String	いいえ	256 MB	このパラメーターは、ディメンションテーブルのルックアップと `lookup` チェンジログプロデューサーの両方のキャッシュサイズを制御します。
merge-engine	同じプライマリキーを持つレコードをマージするためのメカニズム。	String	いいえ	deduplicate	有効な値： deduplicate：最新のレコードのみを保持します。 partial-update：最新のレコードの非 NULL 値で既存のレコードを更新します。他の列は変更されません。 aggregation：指定された集計関数を使用して事前集計を実行します。マージエンジンの詳細な分析については、「Merge Engine」をご参照ください。
partial-update.ignore-delete	削除 (-D) メッセージを無視するかどうかを指定します。	Boolean	いいえ	false	有効な値： true：削除メッセージを無視します。 false：削除メッセージを処理します。 `IllegalStateException` または `IllegalArgumentException` エラーを防ぐために、sequence.field などのパラメーターを使用して削除を処理する戦略を設定する必要があります。説明 Realtime Compute for Apache Flink 8.0.6 以前では、このパラメーターは `merge-engine = 'partial-update'` の部分更新シナリオでのみ有効です。 Realtime Compute for Apache Flink 8.0.7 以降では、このパラメーターは部分更新以外のシナリオとも互換性があり、`ignore-delete` パラメーターと同じ機能です。 `ignore-delete` を代わりに使用することを推奨します。このパラメーターを有効にするかどうかは、ビジネス要件と削除メッセージが想定されているかどうかに基づいて決定してください。削除メッセージがジョブの意図したセマンティクスと一致しない場合は、早期に失敗させる方が多くの場合望ましいです。
ignore-delete	削除 (-D) メッセージを無視するかどうかを指定します。	Boolean	いいえ	false	有効な値は partial-update.ignore-delete と同じです。説明このパラメーターは、Realtime Compute for Apache Flink 8.0.7 以降でのみサポートされます。このパラメーターは `partial-update.ignore-delete` と同じ機能です。 `ignore-delete` を使用し、両方のパラメーターを同時に設定しないことを推奨します。
partition.default-name	デフォルトのパーティション名。	String	いいえ	__DEFAULT_PARTITION__	パーティション列の値が NULL または空文字列の場合に使用するパーティション名。
partition.expiration-check-interval	システムが期限切れのパーティションをチェックする頻度。	String	いいえ	1 h	詳細については、「自動パーティション有効期限の設定方法」をご参照ください。
partition.expiration-time	パーティションが期限切れになるまでの期間。	String	いいえ	なし	パーティションの経過時間がこの値を超えると、パーティションは期限切れになります。デフォルトでは、パーティションは期限切れになりません。システムはパーティション値からパーティションの経過時間を計算します。詳細については、「自動パーティション有効期限の設定方法」をご参照ください。
partition.timestamp-formatter	時間文字列をタイムスタンプに変換するためのフォーマット文字列。	String	いいえ	なし	パーティション値からパーティションの経過時間を抽出するためのフォーマットを指定します。詳細については、「自動パーティション有効期限の設定方法」をご参照ください。
partition.timestamp-pattern	パーティション値を時間文字列に変換するためのフォーマット文字列。	String	いいえ	なし	パーティション値から時間文字列を抽出するためのパターンを指定します。詳細については、「自動パーティション有効期限の設定方法」をご参照ください。
scan.bounded.watermark	スキャンの終了を通知するウォーターマーク値。ソーステーブルは、そのウォーターマークがこの値を超えるとデータの生成を停止します。	Long	いいえ	なし	該当なし
scan.mode	Paimon ソーステーブルの消費位置を指定します。	String	いいえ	default	詳細については、「Paimon ソーステーブルの消費位置の設定方法」をご参照ください。
scan.snapshot-id	Paimon ソーステーブルが消費を開始するスナップショットを指定します。	Integer	いいえ	なし	詳細については、「Paimon ソーステーブルの消費位置の設定方法」をご参照ください。
scan.timestamp-millis	Paimon ソーステーブルが消費を開始する時点を指定します。	Integer	いいえ	なし	詳細については、「Paimon ソーステーブルの消費位置の設定方法」をご参照ください。
snapshot.num-retained.max	保持する最新のスナップショットの最大数。	Integer	いいえ	2147483647	この条件または `snapshot.time-retained` 条件のいずれかが満たされ、かつ `snapshot.num-retained.min` 条件も満たされた場合に、スナップショットの期限切れ処理がトリガーされます。
snapshot.num-retained.min	保持する最新のスナップショットの最小数。	Integer	いいえ	10	該当なし
snapshot.time-retained	スナップショットの保持期間。	String	いいえ	1 h	この条件または `snapshot.num-retained.max` 条件のいずれかが満たされ、かつ `snapshot.num-retained.min` 条件も満たされた場合に、スナップショットの期限切れ処理がトリガーされます。
write-mode	Paimon テーブルの書き込みモード。	String	いいえ	change-log	有効な値： change-log： Paimon テーブルは、プライマリキーに基づいて挿入、削除、更新操作をサポートします。 append-only： Paimon テーブルは挿入操作のみを受け付け、プライマリキーをサポートしません。このモードは、change-log モードよりも効率的です。書き込みモードの詳細については、「Write Mode」をご参照ください。
scan.infer-parallelism	Paimon ソーステーブルの並列度を自動的に推論するかどうかを指定します。	Boolean	いいえ	true	有効な値： true：バケット数に基づいて Paimon ソーステーブルの並列度を自動的に推論します。 false： Realtime Compute for Apache Flink で設定されたデフォルトの並列度を使用します。エキスパートモードが有効な場合、ジョブは代わりにユーザーが設定した並列度を使用します。
scan.parallelism	Paimon ソーステーブルの並列度。	Integer	いいえ	なし	説明ジョブの [デプロイ詳細] > > [リソース設定] タブで [リソースモード] が [エキスパートモード] に設定されている場合、このパラメーターは無視されます。
sink.parallelism	Paimon シンクテーブルの並列度。	Integer	いいえ	なし	説明ジョブの [デプロイ詳細] > > [リソース設定] タブで [リソースモード] が [エキスパートモード] に設定されている場合、このパラメーターは無視されます。
sink.clustering.by-columns	Paimon シンクテーブルへの書き込み時にクラスタリングに使用する列を指定します。	String	いいえ	なし	Paimon の追加のみのテーブル (プライマリキーのないテーブル) の場合、このパラメーターはバッチジョブでのクラスター書き込みを有効にします。このプロセスは、指定された列でデータをグループ化することにより、クエリのパフォーマンスを向上させます。複数の列名はコンマ (,) で区切ります。たとえば、`'col1,col2'` です。クラスタリングの詳細については、「Apache Paimon 公式ドキュメント」をご参照ください。
sink.delete-strategy	システムが撤回メッセージ (-D/-U) を正しく処理することを確実にするための検証戦略を指定します。	Enum	いいえ	NONE	撤回メッセージを処理する際のシンクオペレーターの有効な値と期待される動作： NONE (デフォルト)：検証を実行しません。 IGNORE_DELETE：シンクオペレーターは -U および -D メッセージを無視する必要があります。撤回は発生しません。 NON_PK_FIELD_TO_NULL：シンクオペレーターは -U メッセージを無視する必要があります。 -D メッセージの場合、プライマリキーを保持し、他のすべての非プライマリキーフィールドを NULL に設定します。これは主に、複数のシンクが同じテーブルに書き込む際の部分更新に使用されます。 DELETE_ROW_ON_PK：シンクオペレーターは -U メッセージを無視する必要がありますが、-D メッセージに対してはプライマリキーに対応する行を削除します。 CHANGELOG_STANDARD: シンク演算子は、-U メッセージと -D メッセージのどちらの場合でも、プライマリキーに対応する行を削除する必要があります。説明このパラメーターは、Realtime Compute for Apache Flink 8.0.8 以降でのみサポートされます。撤回時の実際のシンクの動作は、`ignore-delete` や `merge-engine` などの他のパラメーターによって決定されます。このパラメーターは、実際の動作が選択された戦略と一致するかどうかを検証するだけです。不一致がある場合、ジョブは失敗し、修正を提案するエラーメッセージが表示されます。

説明

設定オプションの詳細については、「Apache Paimon 公式ドキュメント」をご参照ください。

機能の詳細

データの鮮度と一貫性

Paimon シンクテーブルは、2 フェーズコミットプロトコルを使用して、各 Flink ジョブのチェックポイント時にデータをコミットします。そのため、データの鮮度は Flink ジョブのチェックポイント間隔によって決まります。各コミットでは、最大 2 つのスナップショットが生成されます。

2 つの Flink ジョブが同じ Paimon テーブルに同時に書き込む際、ジョブが異なるバケットに書き込む場合はシリアライズ可能な一貫性を実現できます。同じバケットに書き込む場合は、スナップショット分離のみが実現されます。この場合、テーブルのデータは両方のジョブの結果が混在する可能性がありますが、データ損失は発生しません。

マージエンジン

Paimon シンクテーブルが同じプライマリキーを持つ複数のレコードを受信すると、一意性を維持するためにそれらを 1 つのレコードにマージします。この動作は、merge-engine パラメーターを設定することで制御できます。次の表に、利用可能なマージエンジンを示します。

マージエンジン	説明
Deduplicate	Deduplicate はデフォルトのエンジンです。同じプライマリキーを持つ複数のレコードの場合、Paimon シンクテーブルは最新のレコードのみを保持し、その他を破棄します。説明最新のレコードが削除メッセージの場合、そのプライマリキーを持つすべてのレコードが破棄されます。
部分更新	部分更新エンジンを使用すると、複数のメッセージで段階的に更新することで、完全なレコードを構築できます。同じプライマリキーを持つ新しいレコードが到着すると、その非 NULL 値が既存のレコードの対応するフィールドを上書きします。このエンジンは、新しいレコードで null であるフィールドを無視し、既存の値を保持します。たとえば、Paimon シンクテーブルが次の 3 つのレコードを順番に受信したとします。 <1, 23.0, 10, NULL> <1, NULL, NULL, 'This is a book'> <1, 25.2, NULL, NULL> 最初の列がプライマリキーの場合、最終的にマージされたレコードは <1, 25.2, 10, 'This is a book'> になります。説明部分更新の結果をストリーム読み取りするには、`changelog-producer` パラメーターを `lookup` または `full-compaction` に設定する必要があります。部分更新エンジンは削除メッセージを処理できません。`partial-update.ignore-delete` パラメーターを `true` に設定して、削除メッセージを無視できます。
Aggregation	一部のユースケースでは、レコードの集計値のみが必要な場合があります。Aggregation エンジンは、指定した集計関数を使用して、同じプライマリキーを持つレコードを集約します。各非プライマリキー列に対して、`fields.<field-name>.aggregate-function` オプションを使用して集計関数を指定する必要があります。指定しない場合、列はデフォルトで `last_non_null_value` 集計関数を使用します。たとえば、次の Paimon テーブル定義について考えます。 `CREATE TABLE MyTable ( product_id BIGINT, price DOUBLE, sales BIGINT, PRIMARY KEY (product_id) NOT ENFORCED ) WITH ( 'merge-engine' = 'aggregation', 'fields.price.aggregate-function' = 'max', 'fields.sales.aggregate-function' = 'sum' );` `price` 列は `max` 関数を使用して集計され、`sales` 列は `sum` 関数を使用して集計されます。2 つの入力レコード <1, 23.0, 15> と <1, 30.2, 20> がある場合、最終結果は <1, 30.2, 35> になります。サポートされている集計関数と対応するデータ型は、次のとおりです。 sum：DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE をサポートしています。 min および max：DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ をサポートしています。 last_value および last_non_null_value：すべてのデータ型をサポートしています。 listagg：STRING をサポートしています。 bool_and および bool_or：BOOLEAN をサポートしています。説明 `sum` 関数のみがリトラクションと削除をサポートしており、その他の集計関数はサポートしていません。これらの関数でリトラクションメッセージを無視するには、`'fields.${field_name}.ignore-retract'='true'` を設定します。集計の結果をストリーム読み取りするには、`changelog-producer` パラメーターを `lookup` または `full-compaction` に設定する必要があります。

チェンジログプロデューサー

changelog-producer パラメーターを設定して、Paimon が任意の入力ストリームに対して完全なチェンジログ (すべての update_after レコードに対応する update_before レコードが存在する) を生成するように構成します。次の表に、利用可能なチェンジログプロデューサーを示します。詳細については、Apache Paimon 公式ドキュメントをご参照ください。

プロデューサー	説明
None	`changelog-producer` を `none` (デフォルト) に設定すると、ダウンストリームの Paimon ソーステーブルでは、特定のプライマリキーのデータの最新の状態のみが表示されます。この不完全なチェンジログでは、コンシューマーがデータの以前の状態を判断できず、削除されたかどうか、または最新の状態が何であるかしかわからないため、正しい計算を実行することが困難になります。たとえば、ダウンストリームコンシューマーが列の合計を計算する必要があり、最新の値が 5 であることしか確認できない場合、合計をどのように更新するかを判断できません。以前の値が 4 の場合、合計は 1 増加する必要があります。以前の値が 6 の場合、合計は 1 減少する必要があります。`update_before` レコードの影響を受けるコンシューマーは `none` プロデューサーを使用すべきではありませんが、他のチェンジログプロデューサーではパフォーマンスオーバーヘッドが発生します。説明データベースなどのダウンストリームコンシューマーが `update_before` データの影響を受けない場合は、`none` プロデューサーを使用できます。特定の要件に基づいてチェンジログプロデューサーを設定してください。
Input	`changelog-producer` を `input` に設定すると、シンクテーブルは入力ストリームをチェンジログファイルに二重に書き込みます。このプロデューサーは、入力ストリーム自体がすでに完全なチェンジログである場合 (変更データキャプチャ (CDC) からのデータなど) にのみ使用してください。
Lookup	`changelog-producer` を `lookup` に設定すると、シンクテーブルはディメンションテーブルルックアップに似たポイントクエリメカニズムを使用して、コミットされる前の現在のスナップショットの完全なチェンジログを生成します。このプロデューサーは、任意の入力ストリームから完全なチェンジログを生成します。 `full-compaction` プロデューサーと比較して、`lookup` プロデューサーはチェンジログの適時性に優れていますが、全体的なリソース消費は多くなります。このオプションは、データ鮮度の要件が高い (分単位など) ユースケースに使用します。
フルコンパクション	`changelog-producer` を `full-compaction` に設定すると、シンクテーブルは各フルコンパクション中に完全なチェンジログを生成します。このプロデューサーは、任意の入力ストリームから完全なチェンジログを生成します。フルコンパクションの間隔は、`full-compaction.delta-commits` パラメーターで指定されます。 `lookup` プロデューサーと比較して、`full-compaction` プロデューサーはレイテンシーが高くなりますが、既存のフルコンパクションプロセスを活用するため、追加の計算は不要です。これにより、全体的なリソース消費が少なくなります。このオプションは、データの鮮度要件が低い (時間単位など) ユースケースに使用します。

書き込みモード

Paimon テーブルは、次の書き込みモードをサポートしています。

モード

説明

Change-log

change-log 書き込みモードは、Paimon テーブルのデフォルトです。このモードは、プライマリキーに基づく挿入、削除、および更新操作をサポートしています。このモードでは、マージエンジンとチェンジログプロデューサーも使用できます。

アペンドオンリー

append-only 書き込みモードは、データの挿入のみをサポートしており、プライマリキーを使用しません。このモードは change-log モードよりも効率的であり、中程度のデータ鮮度 (分単位など) が求められるシナリオでメッセージキューの代替として使用できます。

append-only 書き込みモードの詳細については、Apache Paimon 公式ドキュメントをご参照ください。このモードを使用する場合は、次の点に注意してください。

要件に応じて bucket-key パラメーターを設定してください。設定しない場合、Paimon テーブルはすべての列の値に基づいてデータをバケット化するため、計算効率が低下します。
append-only 書き込みモードは、ある程度データの出力順序を保証できます。具体的な出力順序は次のとおりです。
1. 異なるパーティションからのレコードの場合： scan.plan-sort-partition パラメーターが設定されている場合、値が小さいパーティションからのレコードが最初に出力されます。設定されていない場合は、先に作成されたパーティションからのレコードが最初に出力されます。
2. 同じパーティションおよびバケットからのレコードの場合、先に書き込まれたレコードが最初に出力されます。
3. 同じパーティションだが異なるバケットからのレコードの場合、異なるバケットは異なる並行タスクによって処理されるため、出力順序は保証されません。

CTAS および CDAS のターゲット

Paimon テーブルは、単一テーブルまたはデータベース全体のリアルタイムデータ同期をサポートしています。アップストリームテーブルのスキーマ変更も、Paimon テーブルにリアルタイムで同期されます。詳細については、「Paimon テーブルの管理」および「Paimon カタログの管理」をご参照ください。

Variant 読み取りプルーニング

クエリによって参照される Variant フィールドのみを読み取り、I/O およびメモリオーバーヘッドを削減します。

有効化方法

パラメーター	デフォルト値	説明
`variant.read.pushdown.enabled`	false	`true` に設定して、Variant 読み取りプルーニングを有効にします。

前提条件

Variant データは、現在のバージョンで書き込まれたものである必要があります。以前のバージョンまたは他のエンジンによって書き込まれた Variant データは、読み取りプルーニングをサポートしていません。

サポートされるシナリオ

読み取りプルーニングは、SQL が文字列キーで Variant フィールドにアクセスする場合に有効です。例：

SELECT v['a']            FROM t;   -- 単一レベル
SELECT v['a']['b']       FROM t;   -- ネスト
SELECT v['a'], v['b']    FROM t;   -- 複数のフィールド
SELECT id, v['a']        FROM t;   -- 通常の列と混在
SELECT v['a'] + 1        FROM t;   -- 式で参照

サポートされないシナリオ

Variant 全体を直接参照する場合 (SELECT v FROM t など)。
直接参照とフィールドアクセスを組み合わせる場合 (SELECT v, v['a'] FROM t など)。
配列インデックスアクセス (v[0] など)。
テーブルスキーマにネストされた Row フィールドが含まれている場合。この場合、読み取りプルーニングはテーブル内のどの Variant フィールドにも適用されません。

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

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

デフォルトでは、ルックアップでは、各並列インスタンスがすべてのデータをロードします。このアプローチは、小さいディメンションテーブルにのみ適しています。大きいディメンションテーブルの場合は、以下で説明するシャッフルルックアップの方法を使用してください。

パーティション化されたディメンションテーブル

ディメンションテーブルがパーティション化されていて、最新のパーティション 1 つまたは 2 つのデータのみが必要な場合は、動的パーティションロード機能を使用できます。

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

パラメーター	データ型	デフォルト値	説明
lookup.dynamic-partition	String	該当なし	`max_pt()`：最新のパーティションのみをロードします。`max_two_pt()`：最新の 2 つのパーティションのみをロードします。
lookup.dynamic-partition.refresh-interval	期間	1 h	システムがディメンションテーブルのパーティション更新をチェックする間隔。

大きいディメンションテーブル：固定バケットテーブル

VVR 8.0.8 以降でのみサポートされます。固定バケットテーブル (bucket > 0) の場合、シャッフルルックアップを使用してバケットキーによってデータを並列インスタンスに分散させ、各インスタンスが割り当てられたバケットのデータのみをロードするようにできます。

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

説明

結合キーはバケットキーでなければなりません。バケットキーのデフォルトはプライマリキーです。
この機能は、固定バケットテーブル (bucket > 0) でのみサポートされます。

大きいディメンションテーブル：非固定バケットテーブル

VVR 8.0.10 以降でのみサポートされます。動的バケットテーブルまたはアペンドテーブルの場合、SHUFFLE_HASH または REPLICATED_SHUFFLE_HASH を使用することで、各並列インスタンスがすべてのデータを読み取り、必要な部分のみを保持するようにできます。

-- シャッフルハッシュ
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- レプリケートシャッフルハッシュ
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

SHUFFLE_HASH と REPLICATED_SHUFFLE_HASH の詳細については、「ディメンションテーブルのJOIN文」をご参照ください。

データインジェスト

データインジェスト YAML ジョブでは、Paimon コネクタをシンクとして使用できます。

構文

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

パラメーター

パラメーター	説明	必須	タイプ	デフォルト	備考
type	コネクタのタイプです。	はい	STRING	なし	値は `paimon` である必要があります。
name	シンクの名前です。	いいえ	STRING	なし
catalog.properties.metastore	Paimon カタログのタイプです。	いいえ	STRING	filesystem	有効な値： filesystem (デフォルト) rest (Data Lake Formation (DLF) のみサポート、DLF レガシーはサポート対象外)
catalog.properties.*	Paimon カタログを作成するためのパラメーターです。	いいえ	STRING	なし	詳細については、「Paimon カタログの管理」をご参照ください。
table.properties.*	Paimon テーブルを作成するためのパラメーターです。	いいえ	STRING	なし	詳細については、「Paimon テーブルオプション」をご参照ください。
catalog.properties.warehouse	ファイルストレージのルートディレクトリです。	いいえ	STRING	なし	このパラメーターは、`catalog.properties.metastore` が `filesystem` に設定されている場合にのみ適用されます。
commit.user-prefix	データファイルをコミットするためのユーザー名プレフィックスです。	いいえ	STRING	なし	説明異なるジョブに対して異なるユーザー名を設定することを推奨します。これにより、コミット競合を引き起こしたジョブを特定しやすくなります。
partition.key	パーティションテーブルのパーティションキーです。	いいえ	STRING	なし	異なるテーブルは `;` で区切り、異なるフィールドは `,` で区切り、テーブルとフィールドは `:` で区切ります。例：`testdb.table1:id1,id2;testdb.table2:name`
sink.cross-partition-upsert.tables	プライマリキーにすべてのパーティションキーが含まれていない場合に、クロスパーティションアップサートが必要なテーブルを指定します。	いいえ	STRING	なし	形式：セミコロン `;` を使用してテーブル名を区切ります。パフォーマンス上の推奨事項：この操作はリソースを大量に消費します。これらのテーブル用に個別のジョブを作成してください。重要条件を満たすすべてのテーブルを指定する必要があります。テーブル名を省略すると、データの重複が発生します。
sink.commit.parallelism	Commit オペレーターの並列度を指定します。	いいえ	INTEGER	なし	Commit オペレーターがボトルネックになっている場合、このパラメーターを使用してその並列度を増やし、パフォーマンスを向上させます。このパラメーターは、Realtime Compute for Apache Flink 11.6 以降でのみサポートされます。説明このパラメーターを設定すると、オペレーターの並列度が変更されます。ステートフルジョブを再起動する場合、ジョブが部分的なオペレーター状態を無視できるように、`AllowNonRestoredState` を指定する必要があります。

既存カタログの再利用

Realtime Compute for Apache Flink 11.5 以降では、Flink CDC データインジェストジョブで、データ管理ページから組み込みの Paimon カタログを直接参照できます。これにより、手動での設定を削減できます。

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

データ取り込みジョブは、Paimon カタログのすべてのパラメーターを自動的に再利用できます。これは、YAML ジョブで catalog.properties. というプレフィックスが付いたパラメーターを手動で設定することと同等です。

自動的に再利用されるパラメーターを上書きするには、YAML ジョブで明示的に設定します。明示的な YAML 設定が優先されます。たとえば、前述のサンプルでは、fs.oss.endpoint パラメーターは YAML ジョブの値を使用し、paimon_dlf_catalog の値を上書きします。

例

Paimon をデータインジェストシンクとして使用する場合は、次の例を参照して、Paimon カタログのタイプに基づいてジョブを設定してください。

filesystem Paimon カタログを使用して Object Storage Service (OSS) に書き込む場合の設定例：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx

catalog.properties プレフィックスが付いたパラメーターについては、Paimon Filesystem カタログの作成をご参照ください。

rest Paimon カタログを使用してData Lake Formation (DLF) に書き込む場合の設定例：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (オプション) 削除ベクトルを有効にして、読み取りパフォーマンスを向上させます。
  table.properties.deletion-vectors.enabled: true

catalog.properties プレフィックスが付いたパラメーターについては、Flink CDC カタログ設定パラメーターをご参照ください。

スキーマ変更

データインジェストシンクとして使用する場合、Paimon は次のスキーマ変更イベントをサポートします。

CREATE TABLE EVENT
ADD COLUMN EVENT
ALTER COLUMN TYPE EVENT (プライマリキー列のデータ型の変更はサポートされていません。)
RENAME COLUMN EVENT
DROP COLUMN EVENT
TRUNCATE TABLE EVENT
DROP TABLE EVENT

説明

下流の Paimon テーブルがすでに存在する場合、ジョブは既存のスキーマに書き込み、テーブルを再度作成しようとはしません。

Realtime Compute for Apache Flink:Paimon コネクタ