Data Lake Formation:DLF への Flink CDC 接続

MySQL データベース全体から DLF へのデータインジェスト

以下のジョブは、MySQL データベース内のすべてのテーブルを DLF に同期します。

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (オプション) 増分フェーズ中に作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (オプション) 潜在的な TaskManager メモリ不足 (OOM) エラーを防ぐために、無制限シャードを優先します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 一致するテーブルからのみデータを逆シリアル化して読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (オプション) コミットユーザー。競合を避けるため、異なるジョブには異なるコミットユーザーを設定してください。
  commit.user: your_job_name
  # (オプション) 読み取りパフォーマンスを向上させるために削除ベクターを有効にします。
  table.properties.deletion-vectors.enabled: true

推奨される MySQL ソース オプションは次のとおりです。

パーティションテーブルへのデータインジェスト

非パーティションソーステーブルから DLF のパーティションテーブルにインジェストするには、変換モジュールに partition-keys オプションを追加します。詳細については、「Flink CDC を使用したデータインジェスト」をご参照ください。

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (オプション) 増分フェーズ中に作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (オプション) 潜在的な TaskManager メモリ不足 (OOM) エラーを防ぐために、無制限シャードを優先します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 一致するテーブルからのみデータを逆シリアル化して読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (オプション) コミットユーザー。競合を避けるため、異なるジョブには異なるコミットユーザーを設定してください。
  commit.user: your_job_name
  # (オプション) 読み取りパフォーマンスを向上させるために削除ベクターを有効にします。
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    partition-keys: id,pt
  - source-table: mysql_test.tbl2
    partition-keys: id,pt

追記専用テーブルへのデータインジェスト

インジェスト中に論理削除 (ソフトデリート) を実装するには、変換モジュールで converter-after-transform: SOFT_DELETE を使用します。これにより、削除操作が挿入操作に変換され、ダウンストリームテーブルはすべての変更操作を完全に記録します。projection 内の __data_event_type__ フィールドは、変更タイプを新しい列としてダウンストリームテーブルに書き込みます。

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (オプション) 増分フェーズ中に作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (オプション) 潜在的な TaskManager メモリ不足 (OOM) エラーを防ぐために、無制限シャードを優先します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 一致するテーブルからのみデータを逆シリアル化して読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (オプション) コミットユーザー。競合を避けるため、異なるジョブには異なるコミットユーザーを設定してください。
  commit.user: your_job_name
  # (オプション) 読み取りパフォーマンスを向上させるために削除ベクターを有効にします。
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    partition-keys: id,pt
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE
  - source-table: mysql_test.tbl2
    partition-keys: id,pt
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

詳細については、「Flink CDC を使用したデータインジェスト」をご参照ください。

Kafka から DLF へのリアルタイムデータ同期

以下のジョブは、Kafka inventory トピック (Debezium JSON 形式のテーブル customers と products) から CDC データを読み取り、対応する送信先テーブルに DLF で同期します。Debezium JSON メッセージにはプライマリキー情報が含まれていないため、プライマリキーは変換モジュールで明示的に指定されます。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (オプション) コミットユーザー。競合を避けるため、異なるジョブには異なるコミットユーザーを設定してください。
  commit.user: your_job_name
  # (オプション) 読み取りパフォーマンスを向上させるために削除ベクターを有効にします。
  table.properties.deletion-vectors.enabled: true

# Debezium JSON にはプライマリキー情報が含まれていないため、プライマリキーを明示的に指定します。
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Kafka ソースに関する追加の注意事項:

• Kafka ソースは、canal-json、debezium-json (デフォルト)、および json の 3 つのデータ形式をサポートしています。

• 複数の Kafka パーティションから DLF の単一テーブルにインジェストする場合、debezium-json.distributed-tables または canal-json.distributed-tables を true に設定します。

• Kafka ソースは、schema.inference.strategy オプションを介して複数のスキーマ推論ポリシーをサポートしています。詳細については、「Message Queue for Apache Kafka」をご参照ください。

詳細については、「Flink CDC を使用したデータインジェスト」をご参照ください。