Flink CDC Transform モジュール - Realtime Compute for Apache Flink

Transform ルールのパラメーター

Transform モジュールは、データ列の直接的な操作をサポートしています。データ同期中に、既存の列を削除または拡張したり、不要なデータをフィルタリングしたりできます。以下のパラメーターを使用して、Transform ルールを定義します。

パラメーター	説明	必須	注意
`source-table`	変換対象の上流テーブルを指定します。	はい	正規表現をサポートしています。
`projection`	上流テーブルのプロジェクションルールを指定します。これにより、変換後のすべてのデータ列が決定されます。	いいえ	構文は SQL の SELECT 文に似ています。このパラメーターが指定されていない場合、列の追加や削除は行われません。詳細については、「データスクリーニング」をご参照ください。利用可能な組み込み関数の詳細については、「Flink CDC 組み込み関数」をご参照ください。
`filter`	行フィルタリングのルールです。	いいえ	構文は SQL の WHERE 文に似ています。このパラメーターが指定されていない場合、行のフィルタリングは行われません。
`primary-keys`	変換後のプライマリキー列を指定します。	いいえ	このパラメーターが指定されていない場合、元のスキーマのプライマリキー定義は保持されます。プライマリキーはコンマ (`,`) で区切られます。重要デフォルトでは、アップストリームから主キー列を削除できません。アップストリームの主キー列の数を減らすには、`transform.allow.trimming.pk-columns` パラメーターを `pipeline:` 設定に追加します。カスタムのプライマリキー列を定義する場合、上流テーブルからの入力データが PRIMARY KEY 制約に準拠していることを確認してください。クロスパーティション書き込み時のデータ順序の乱れの問題を防ぐため、上流テーブルのプライマリキー列をカスタムプライマリキー列に含めることを推奨します。
`partition-keys`	変換後のパーティションキー列のリストを指定します。	いいえ	このパラメーターが指定されていない場合、元のスキーマのパーティションキー定義が保持されます。パーティションキーは、コンマ (`,`) で区切られます。重要カスタムのパーティションキー列を定義する場合、クロスパーティション書き込み時にデータ順序の乱れが発生する可能性があります。データ順序を保証するには、primary-keys の設定も合わせて考慮することを推奨します。
`table-options`	シンクに追加の設定情報を渡します。	いいえ	Paimon シンクのバケット数やコメントなど、オプションのプロパティのリストです。設定項目はカンマ (`,`) で区切られます。各項目のキーと値は等号 (`=`) で区切られます。設定例： `key1=value1,key2=value2`
`description`	Transform ルールの説明です。	いいえ	なし。
`converter-after-transform`	変換が完了したデータに追加処理を行うコンバーターです。	いいえ	詳細については、「Transform 後のコンバーター」をご参照ください。

注意事項

Transform モジュール内のステートメントを変更した後は、ステートレスな初期状態からジョブを再起動する必要があります。
通常、projection と filter のステートメントは引用符で囲む必要はありません。
```
transform:
  - projection: a, b, c
    # 以下と等価
  - projection: "a, b, c"
```
しかし、射影式が * や ' などの特殊文字で始まる場合、式全体が有効な YAML 文字列リテラルとして解析されない可能性があります。この場合、手動で式全体を単一引用符 (') または二重引用符 (") で囲むか、バックスラッシュ (\) を使用して文字をエスケープする必要があります。
```
transform:
  - projection: *, 42      # 無効な YAML
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK
```
Ververica Runtime (VVR) 11.6 以降では、上流から推測されたプライマリキー設定をクリアして、テーブルをプライマリキーなしに変換できます。構文は次のとおりです。
```
transform:
  - source-table: db.tbl
    primary-keys: '' # db.tbl テーブルの PRIMARY KEY 設定をクリアします
```

データスクリーニング

Transform モジュールは、SQL に似た構文を使用してデータスクリーニング (プロジェクション) ルールを定義します。これらのルールにより、特定の列を選択したり、計算列やメタデータ列を追加したりできます。

列のプルーニング

ソーステーブルから特定の列を抽出し、それらを下流に同期するには、プロジェクションルールで目的の列を指定します。指定されていない列は下流に送信されません。

transform:
  - source-table: db.tbl
    projection: col_1, col_3, col_4 # col_2 はプルーニングされます

重要

列をプルーニングすると、上流テーブルのスキーマが変更された場合に、上流と下流のテーブルスキーマに不整合が生じる可能性があります。

ワイルドカード文字

後で追加される新しいカラムも含めて、ソーステーブルのすべてのカラムを変更せずにダウンストリームに送信するには、射影ルールでワイルドカード文字のアスタリスク (*) を使用できます。

説明

プロジェクションルールがワイルドカード文字 (*) を使用しない場合、結果のスキーマは固定されます。スキーマはプロジェクションルールで指定されたバージョンと一貫性が保たれ、スキーマエボリューションは同期できません。

たとえば、*, 'extras' AS extras は、アップストリームスキーマの列の後に追加の列が付加されることを示します。アップストリームスキーマの進化は、引き続きダウンストリームに同期されます。

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

計算列

プロジェクションルールで <Expression> AS <ColName> 構文を使用して、計算列を追加できます。式は、アップストリームの各レコードに対して評価され、その結果が対応する列に格納されます。

説明

計算列の式では、たとえ参照先の列が先に出現していても、別の計算列の値を参照することはできません。たとえば、a, b AS c, c AS d は有効な式ではありません。

たとえば、アップストリームテーブル db.tbl がデータレコード [+I, id = 1] を送信すると、レコードはデータ行 [+I, id = 1, inc_id = 2] に変換されてダウンストリームに送信されます。

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

メタデータ列

プロジェクションルールを作成する際、以下の定義済みメタデータ列を通常のデータ列と同じように使用できます。

重要

通常のデータ列をメタデータ列と同じ名前で定義しないでください。

以下のメタデータ列は、すべてのコネクターに適用されます。

メタデータ列名	データ型	説明
`__namespace_name__`	String	この変更レコードのソーステーブルの名前空間名です。
`__schema_name__`	String	この変更レコードのソーステーブルのスキーマ名です。
`__table_name__`	String	この変更レコードのソーステーブルのテーブル名です。
`__data_event_type__`	String	この変更レコードの操作タイプ (`+I`、`-U`、`+U`、または `-D`)。重要 CDC イベントは、更新において更新前のレコードと更新後のレコードを常に 1 つのイベントにまとめます。したがって、同じ更新イベント内では `__data_event_type__` の内容は、それぞれ `-U` および `+U` となります。これをプライマリキーとして使用しないでください。

たとえば、上流テーブルの完全修飾名を計算列に書き込み、それを下流に送信することができます。

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

次の表に、データベースコネクターと名前空間、スキーマ、およびテーブル名のマッピング関係を示します。

データベースタイプ	名前空間名	スキーマ名	テーブル名
MySQL	-	データベース	テーブル
Kafka	-	-	トピック
SLS	-	プロジェクト	LogStore
MongoDB	-	データベース	コレクション
Paimon	-	データベース	テーブル
Hologres	-	スキーマ	テーブル
StarRocks	-	データベース	テーブル
Doris	-	データベース	テーブル
Postgres	データベース説明これは、`table-id.include-database` パラメーターが有効な場合にのみ有効です。	スキーマ	テーブル

データフィルタリング

Transform モジュールは、SQL に似た構文を使用して行フィルタリングルールを定義します。

フィルターで指定するルールは、BOOLEAN 値に評価される式でなければなりません。この式では、ソーステーブルの任意の列や計算列を参照できます。

変更レコードがフィルター付きの変換ルールに一致し、フィルター式が FALSE と評価された場合、そのデータ行はダウンストリームに送信されません。

説明

プロジェクションルールで計算列を使用して既存の上流列を上書きする場合、フィルター式はその計算列を参照します。

たとえば、次の Transform ルールは有効です。

transform:
  - source-table: db.tbl
    projection: CAST(id AS VARCHAR) AS id
    filter: CHAR_LENGTH(id) > 5

filter 式で参照される id は、VARCHAR 型に変換された計算列です。

高度な設定ルール

プルーニングされていない列とメタデータの参照

例 1：プルーニングされた列に基づくフィルター

上流テーブルスキーマが [INT a, INT b, BOOLEAN c] であるとします。カラム a と b を出力し、c が true の行のみを保持するには、次の設定ルールを使用できます。

transform:
  - source-table: ...
    projection: a, b
    filter: c

例 2：メタデータ列に基づくフィルター

メタデータ列は、projection で明示的に定義しなくても、フィルター条件として直接使用できます。たとえば、DELETE 型の変更データを除外できます。

transform:
  - source-table: db.tbl
    projection: a, b
    filter: __data_event_type__ <> '-D'

既存列の上書き

projection 内では、上流カラムと同じ名前のフィールドを定義して、そのカラムの値または型を上書きできます。この方法により、スキーマエボリューションが正しく同期されることが保証されます。

例：強制的な型変換

アップストリームテーブルのスキーマが [INT a, INT b, BOOLEAN c] であるとします。カラム名を変更せずにカラム a を強制的に文字列型に変換するには、以下の設定を使用できます。

transform:
  - source-table: db.tbl
    projection: \*, CAST(a AS VARCHAR) AS a

下流テーブルのスキーマは [STRING a, INT b, BOOLEAN c] になります。元のカラム a は、新しい定義によって上書きされます。

フィルター条件での計算列の再利用

filter は、projection で定義された計算列のエイリアスを直接参照できます。

例：計算結果の参照

projection で新しい列 d を定義して、filter で直接使用できます。

transform:
  - source-table: db.tbl
    projection: a, b, c, a + b + c AS d
    filter: d > 100

この設定は次の設定と同じ効果を持ちますが、より可読性が高くなります。

transform:
  - source-table: ...
    projection: a, b, c, a + b + c AS d
    filter: a + b + c > 100

Transform 後のコンバーター

converter-after-transform は、すべての変換ルールが適用された後にデータ変更を処理するために使用されます。複数のコンバーターをカンマ (,) で区切って指定できます。コンバーターは、記載されている順序で適用されます。以下の設定値がサポートされています。

コンバーター名	機能	サポートバージョン
SOFT_DELETE	削除操作を挿入操作に変換します。	VVR 8.0.11 以降。
FIELD_NAME_LOWER_CASE	すべてのテーブルフィールド名を小文字に変換します。	VVR 11.1 以降。

ソフトデリート

SOFT_DELETE コンバーターは、__data_event_type__ メタデータ列と連携して、論理削除を実装します。これは、データがダウンストリームシステムで物理的に削除されないことを意味します。たとえば、次の変換設定は論理削除を実装します。削除操作は挿入に変換され、対応するデータの op_type は削除を示すために -D に更新されます。

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE