Realtime Compute for Apache Flink:Paimon コネクタ

項目

説明

サポートされるタイプ

ソーステーブル、ディメンションテーブル、結果テーブル、およびデータインジェストシンク

実行モード

ストリーミングモードとバッチモード

データフォーマット

N/A

メトリック

N/A

API タイプ

SQL API とデータインジェスト YAML API

結果テーブルでのデータの更新または削除

サポートされています

オプション

説明

タイプ

必須

デフォルト値

備考

connector

テーブルのタイプ。

String

いいえ

デフォルト値なし

• Apache Paimon カタログで Apache Paimon テーブルを作成する場合、このオプションを指定する必要はありません。

• Apache Paimon 以外のストレージのカタログに一時的な Apache Paimon テーブルを作成する場合は、値を paimon に設定します。

path

テーブルのストレージパス。

String

いいえ

デフォルト値なし

• Apache Paimon カタログで Apache Paimon テーブルを作成する場合、このオプションを指定する必要はありません。

• Apache Paimon 以外のストレージのカタログに一時的な Apache Paimon テーブルを作成する場合は、このオプションをテーブルを保存する HDFS または OSS ディレクトリに設定します。

auto-create

一時的な Apache Paimon テーブルを作成するときに、指定されたパスに Apache Paimon テーブルファイルが存在しない場合に Apache Paimon テーブルファイルを自動的に作成するかどうかを指定します。

Boolean

いいえ

false

有効な値:

• false (デフォルト): 指定されたパスに Apache Paimon テーブルファイルが存在しない場合、エラーが返されます。

• true: 指定されたパスが存在しない場合、システムは自動的に Apache Paimon テーブルファイルを作成します。

bucket

各パーティションのバケット数。

Integer

いいえ

1

Apache Paimon テーブルに書き込まれるデータは、bucket-key オプションで指定された列に基づいて各バケットに分散されます。

bucket-key

バケットキー列。

String

いいえ

デフォルト値なし

Apache Paimon テーブルに書き込まれるデータが異なるバケットに分散される基準となる列。

列名をコンマ (,) で区切ります。たとえば、'bucket-key' = 'order_id,cust_id' は、データが order_id 列と cust_id 列に基づいてバケットに分散されることを示します。

changelog-producer

増分データ生成メカニズム。

String

いいえ

none

Apache Paimon は、下流のデータ消費を容易にするために、任意の入力データストリームに対して完全な増分データを生成できます。各 UPDATE_AFTER データレコードは、UPDATE_BEFORE データレコードに対応します。有効な値:

• none (デフォルト): 増分データは生成されません。下流のコンシューマーは、ストリーミングモードで Apache Paimon テーブルからデータを読み取ることができます。ただし、下流のコンシューマーによって読み取られる増分データには UPDATE_AFTER データのみが含まれ、UPDATE_BEFORE データは含まれません。

• input: 入力データストリームは、デュアルライトモードで増分データとして増分データファイルに書き込まれます。

• full-compaction: 完全なコンパクションが実行されるたびに、完全な増分データが生成されます。

• lookup: コミットセーブポイントが実行される前に、完全な増分データが生成されます。

増分データ生成メカニズムの選択方法の詳細については、このトピックの「増分データ生成メカニズム」セクションをご参照ください。

full-compaction.delta-commits

完全なコンパクションが実行される最大間隔。

Integer

いいえ

デフォルト値なし

コミットセーブポイントの数がこのオプションの値に達すると、完全なコンパクションが確実にトリガーされます。

lookup.cache-max-memory-size

Apache Paimon ディメンションテーブルのメモリキャッシュサイズ。

String

いいえ

256 MB

このオプションの値は、ディメンションテーブルとルックアップチェンジログプロデューサーの両方のキャッシュサイズを決定します。

merge-engine

同じプライマリキーを持つデータをマージするメカニズム。

String

いいえ

deduplicate

有効な値:

• deduplicate: 最新のデータレコードのみが保持されます。

• partial-update: 最新のデータと同じプライマリキーを持つ既存のデータは、null でない列で最新のデータによって上書きされます。他の列のデータは変更されません。

• aggregation: 事前集約を実行するために集計関数が指定されます。

データマージメカニズムの詳細については、このトピックの「データマージメカニズム」セクションをご参照ください。

partial-update.ignore-delete

削除メッセージを無視するかどうかを指定します。

Boolean

いいえ

false

有効な値:

• true: 削除メッセージは無視されます。

• false: 削除メッセージは無視されません。sequence.field または他のオプションを構成して、シンクが削除メッセージをどのように処理するかを設定する必要があります。そうしないと、IllegalStateException や IllegalArgumentException のようなエラーが発生します。

ignore-delete

削除メッセージを無視するかどうかを指定します。

Boolean

いいえ

false

その有効な値は partial-update.ignore-delete と同じです。

partition.default-name

パーティションのデフォルト名。

String

いいえ

__DEFAULT_PARTITION__

パーティションキー列の値が null または空の文字列の場合、このオプションの値がパーティション名として使用されます。

partition.expiration-check-interval

システムがパーティションの有効期限切れをチェックする間隔。

String

いいえ

1h

詳細については、「自動パーティションの有効期限切れを構成するにはどうすればよいですか？」をご参照ください。

partition.expiration-time

パーティションの有効期間。

String

いいえ

デフォルト値なし

パーティションが存在する期間がこのオプションの値を超えると、パーティションは有効期限切れになります。デフォルトでは、パーティションは決して有効期限切れになりません。

パーティションが存在する期間は、パーティションの値に基づいて計算されます。詳細については、「自動パーティションの有効期限切れを構成するにはどうすればよいですか？」をご参照ください。

partition.timestamp-formatter

時間文字列をタイムスタンプに変換するために使用されるパターン。

String

いいえ

デフォルト値なし

このオプションは、パーティション値からパーティションが存在する期間を抽出するために使用されるパターンを指定します。詳細については、「自動パーティションの有効期限切れを構成するにはどうすればよいですか？」をご参照ください。

partition.timestamp-pattern

パーティション値を時間文字列に変換するために使用されるパターン。

String

いいえ

デフォルト値なし

このオプションは、パーティション値からパーティションが存在する期間を抽出するために使用されるパターンを指定します。詳細については、「アップストリームおよびダウンストリームストレージに関するよくある質問」トピックの「自動パーティションの有効期限切れを構成するにはどうすればよいですか？」セクションをご参照ください。

scan.bounded.watermark

有界ストリーミングモードの終了条件。Apache Paimon ソーステーブルのデータのウォーターマークがこのオプションの値を超えると、Apache Paimon ソーステーブルのデータの生成が終了します。

Long

いいえ

デフォルト値なし

N/A。

scan.mode

Apache Paimon ソーステーブルのコンシューマオフセット。

String

いいえ

default

詳細については、「Apache Paimon ソーステーブルのコンシューマオフセットを指定するにはどうすればよいですか？」をご参照ください。

scan.snapshot-id

Apache Paimon ソーステーブルがデータの消費を開始するセーブポイントの ID。

Integer

いいえ

デフォルト値なし

詳細については、「Apache Paimon ソーステーブルのコンシューマオフセットを指定するにはどうすればよいですか？」をご参照ください。

scan.timestamp-millis

Apache Paimon ソーステーブルがデータの消費を開始する時点。

Integer

いいえ

デフォルト値なし

詳細については、「Apache Paimon ソーステーブルのコンシューマオフセットを指定するにはどうすればよいですか？」をご参照ください。

snapshot.num-retained.max

保持できる最新のセーブポイントの最大数。

Integer

いいえ

2147483647

セーブポイントの有効期限切れは、snapshot.num-retained.max オプションまたは snapshot.time-retained オプションで指定された条件が満たされ、かつ snapshot.num-retained.min オプションで指定された条件が満たされた場合にのみトリガーされます。

snapshot.num-retained.min

保持できる最新のセーブポイントの最小数。

Integer

いいえ

10

N/A。

snapshot.time-retained

セーブポイントを保持できる期間。

String

いいえ

1h

セーブポイントの有効期限切れは、snapshot.num-retained.max オプションまたは snapshot.time-retained オプションで指定された条件が満たされ、かつ snapshot.num-retained.min オプションで指定された条件が満たされた場合にのみトリガーされます。

write-mode

Apache Paimon テーブルの書き込みモード。

String

いいえ

change-log

有効な値:

• change-log: データは、プライマリキーに基づいて Apache Paimon テーブルに挿入、削除、および更新されます。

• append-only: Apache Paimon テーブルはデータの挿入のみを許可し、プライマリキーに基づく操作はサポートしません。このモードは、change-log モードよりも効率的です。

書き込みモードの詳細については、「書き込みモード」をご参照ください。

scan.infer-parallelism

Apache Paimon ソーステーブルの並列処理の次数を自動的に推測するかどうかを指定します。

Boolean

いいえ

true

有効な値:

• true: Apache Paimon ソーステーブルの並列処理は、バケット数に基づいて自動的に推測されます。

• false: Ververica Platform (VVP) に基づいて構成されたデフォルトの並列処理の次数が使用されます。リソース構成がエキスパートモードの場合、構成された並列処理の次数が使用されます。

scan.parallelism

Apache Paimon ソーステーブルの並列処理の次数。

Integer

いいえ

デフォルト値なし

sink.parallelism

Apache Paimon 結果テーブルの並列処理。

Integer

いいえ

デフォルト値なし

sink.clustering.by-columns

Apache Paimon 結果テーブルにデータを書き込むために使用されるクラスタリング列。

String

いいえ

デフォルト値なし

プライマリキーのない Apache Paimon append-only テーブルの場合、バッチデプロイメントでこのオプションを指定して、データ書き込みのクラスタリング機能を有効にできます。この機能が有効になると、データはサイズによって特定の列にクラスタリングされて表示されます。これにより、テーブルのクエリ速度が向上します。

複数の列名をコンマ (,) で区切ります。例: 'col1,col2'。

クラスタリング機能の詳細については、「クラスタリング」をご参照ください。

sink.delete-strategy

システムがリトラクション (削除および更新前) メッセージを処理するための検証戦略を指定します。

Enum

いいえ

NONE

有効な値:

• NONE (デフォルト): 検証は実行されません。

• IGNORE_DELETE: シンクオペレーターはリトラクションメッセージを無視する必要があります。

• NON_PK_FIELD_TO_NULL: シンクオペレーターは update_before メッセージを無視する必要があります。delete メッセージを受信した場合、対応するプライマリキー値を保持し、非プライマリキー列の値を削除する必要があります。

  この値は、データが複数のシンクから単一のテーブルに書き込まれる部分更新シナリオに適しています。

• DELETE_ROW_ON_PK: シンクオペレーターは update_before メッセージを無視し、delete メッセージに対応するデータレコードを削除する必要があります。

• CHANGELOG_STANDARD: シンクオペレーターは、受信した update_before および delete メッセージに対応するデータレコードを削除する必要があります。

データマージメカニズム

説明

deduplicate

これは merge-engine オプションのデフォルト値です。データマージメカニズムが deduplicate で、複数のデータレコードが同じプライマリキーを持つ場合、Apache Paimon 結果テーブルは最新のデータレコードのみを保持し、他のデータレコードは破棄します。

partial-update

partial-update メカニズムを使用すると、複数のメッセージを使用してデータを更新し、最終的に完全なデータを取得できます。既存のデータと同じプライマリキーを持つ新しいデータは、既存のデータを上書きします。NULL 値を持つ列は既存のデータを上書きできません。

たとえば、Apache Paimon 結果テーブルは次のデータレコードを順番に受信します。

• <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'> になります。

aggregation

特定のシナリオでは、集約された値のみが必要になる場合があります。集約メカニズムは、指定された集計関数に基づいて同じプライマリキーを持つデータを集約できます。プライマリキーではない各列に集計関数を指定するには、fields.<field-name>.aggregate-function を使用する必要があります。プライマリキーではない列に集計関数を指定しない場合、その列はデフォルトで last_non_null_value 集計関数を使用します。次のサンプルコードは、集約メカニズムを使用して Apache 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 関数に基づいて集約されます。入力データレコードが <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