Data Lake Formation:Iceberg REST を使用した Flink CDC からの DLF カタログへのアクセス

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: iceberg
  using.built-in-catalog: catalog_name

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: iceberg
  using.built-in-catalog: catalog_name

transform:
  - source-table: mysql_test.tbl1
    # (オプション) パーティションキーを設定します。
    partition-keys: id,pt
  - source-table: mysql_test.tbl2
    partition-keys: id,pt

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: iceberg
  using.built-in-catalog: catalog_name

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

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

Kafka の inventory トピックに、customers と products の 2 つのテーブルの変更データが Debezium JSON 形式で保存されているとします。このジョブは、これらのテーブルのデータを対応する DLF ターゲットに同期します。

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: iceberg
  using.built-in-catalog: catalog_name

# Debezium JSON には主キー情報がないため、手動で追加します。
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

リアルタイム Kafka ログの DLF への同期

Kafka クラスターがカスタム JSON 形式でデータを保存している場合、Flink CDC YAML ジョブを設定して DLF に同期できます。システムは、データ型の推論、スキーマ推論、およびスキーマ進化を自動的に処理します。

inventory トピックに、単一のログテーブルのデータが JSON 形式で保存されているとします。このジョブは、そのデータを対応する DLF テーブルに同期します。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # (オプション) JSON データ内のネストされた列を再帰的にフラット化します。
  json.infer-schema.flatten-nested-columns.enable: true
  # (オプション) 最初の 100 件の解析エラーをスキップします。エラーが 100 件を超えるとジョブは失敗します。
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

# テーブルに主キーを追加します。
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

# inventory トピックのすべてのデータを test_database.inventory に書き込みます。
route:
  - source-table: inventory
    sink-table: test_database.inventory

pipeline:
  # (オプション) 処理例外を引き起こしたダーティデータをログに記録します。
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

inventory トピックに、複数のログテーブルのデータが JSON 形式で保存されており、データベース名とテーブル名が databaseName フィールドと tableName フィールドにあるとします。このジョブは、これらのテーブルのデータを対応する DLF ターゲットに同期します。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # (オプション) JSON データ内のネストされた列を再帰的にフラット化します。
  json.infer-schema.flatten-nested-columns.enable: true
  # databaseName と tableName フィールドの値をデータベース名とテーブル名として使用します。
  json.decode.parser-table-id.fields: databaseName,tableName
  # (オプション) 最初の 100 件の解析エラーをスキップします。エラーが 100 件を超えるとジョブは失敗します。
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

# テーブルに主キーを追加します。
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

# ods.inventory、ods.customer、ods.user をそれぞれ test_database.inventory、test_database.customer、test_database.user に書き込みます。
route:
  - source-table: ods.inventory
    sink-table: test_database.inventory
  - source-table: ods.customer
    sink-table: test_database.customer
  - source-table: ods.user
    sink-table: test_database.user

pipeline:
  # (オプション) 処理例外を引き起こしたダーティデータをログに記録します。
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

JSON 形式の Kafka ソーステーブルのスキーマ推論と進化戦略の詳細については、「スキーマ解析と変更同期戦略」をご参照ください。