Realtime Compute for Apache Flink:Flink CDC ジョブの複雑なユースケース

新規空テーブルのホット同期

増分フェーズ中に新規作成された空のテーブルを再起動なしでリアルタイムに同期するには、scan.binlog.newly-added-table.enabled パラメータを設定します。この方法はジョブの再起動を回避できるため推奨されます。

例えば、MySQL のdlf_testデータベースからすべてのテーブルを同期するデータインジェストジョブを実行しているとします。ソースで products という名前の新規の空テーブルが作成されました。ジョブを再起動せずにこの新規テーブルを同期するには、次のようにジョブ設定で scan.binlog.newly-added-table.enabled: true を設定します。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) 増分フェーズ中に新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  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

この設定により、ジョブはdlf_testデータベースのすべての新規テーブルをシンクに自動的に作成します。

履歴データを含むテーブルの同期

MySQL データベースに既にcustomersテーブルとproductsテーブルが含まれているとします。ただし、最初はcustomersテーブルのみを同期するようにジョブを設定していました。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.customers
  server-id: 8601-8604
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  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

ジョブがしばらく実行された後、履歴データを含むデータベースのすべてのテーブルを同期する必要がある場合は、ジョブを再起動する必要があります。次の手順に従ってください。

1. セーブポイントを使用してジョブを停止します。

2. MySQL ソース設定のtablesパラメータを、すべてのテーブルが対象となるように変更します。次に、scan.binlog.newly-added-table.enabled パラメータが存在する場合は削除し、scan.newly-added-table.enabled を追加します。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) 新しく追加されたテーブルの全量および増分データを同期します。
  scan.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  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

3. セーブポイントからジョブを再起動します。

例えば、dlf_test MySQL データベース内の products_tmp テーブルを除くすべてのテーブルを同期するには、以下の設定を使用します。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  # (オプション) 同期したくないテーブルを除外します。
  tables.exclude: dlf_test.products_tmp
  server-id: 8601-8604
  # (オプション) 増分フェーズ中に新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  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

この設定により、Flink CDC データインジェストジョブは dlf_test データベースのすべてのテーブルをシンクに自動的に作成しますが、products_tmp テーブルは除外されます。ジョブは、同期対象テーブルのスキーマとデータをリアルタイムで更新し続けます。

メタデータ列の追加

データを書き込む際に、transform モジュールを使用してメタデータ列を追加できます。たとえば、次の設定では、ダウンストリームテーブルにテーブル名、操作時間、操作タイプが追加されます。詳細については、「transform モジュール」をご参照ください。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) 新しく追加されたテーブルの全量および増分データを同期します。
  scan.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルタを有効にして読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # 操作時刻をメタデータとして含めます。
  metadata-column.include-list: op_ts
transform:
  - source-table: dlf_test.customers
    projection: __schema_name__ || '.' || __table_name__  as identifier, op_ts, __data_event_type__ as op, *
    # (オプション) プライマリキーを変更します。
    primary-keys: id,identifier
    description: add identifier, op_ts and op
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

ソーステーブルには、すべての変更データイベントタイプが含まれます。シンクテーブルでDELETE操作をINSERTに変換してソフトデリートを実装するには、transformモジュールにconverter-after-transform: SOFT_DELETE設定を追加してください。

計算列の追加

データを書き込む際、transformモジュールを使用して計算列を追加できます。例えば、次の設定では、created_atフィールドを変換してdtフィールドを作成し、それをシンクテーブルのパーティションキーとして使用します。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) 新しく追加されたテーブルの全量および増分データを同期します。
  scan.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルタを有効にして読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # 操作時刻をメタデータとして含めます。
  metadata-column.include-list: op_ts
transform:
  - source-table: dlf_test.customers
    projection: DATE_FORMAT(created_at, 'yyyyMMdd') as dt, *
    # (オプション) パーティションキーを設定します。
    partition-keys: dt
    description: add dt
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

シャード化されたテーブルのマージ

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) 新しく追加されたテーブルの全量および増分データを同期します。
  scan.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルタを有効にして読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  # dlf_test データベースから product_[0-9]+ パターンに一致するすべてのテーブルを dlf.products テーブルにマージします。
  - source-table: dlf_test.product_[0-9]+
    sink-table: dlf.products
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

データベース同期

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) 新しく追加されたテーブルの全量および増分データを同期します。
  scan.newly-added-table.enabled: true
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルタを有効にして読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  # dlf_test データベースのすべてのテーブルを dlf データベースに同期し、テーブル名部分に ods_ プレフィックスを付けて名前を変更します。
  - source-table: dlf_test.\.*
    sink-table: dlf.ods_<>
    replace-symbol: <>
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 および Kafka ソースに対応しています。対応するスイッチを有効にし、特定の開始日時を設定します。

ジョブパラメータの設定

ジョブ定義では、パラメータを設定してソーステーブルの開始時刻を指定できます。

例えば、MySQL ソースの場合、ジョブ設定で scan.startup.mode: timestamp を設定することで、特定のタイムスタンプから開始できます。次に設定例を示します。

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (オプション) タイムスタンプモードでジョブを開始します。
  scan.startup.mode: timestamp
  # このモードで起動タイムスタンプを指定します。
  scan.startup.timestamp-millis: 1667232000000
  # (オプション) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限チャンクを優先的にディスパッチし、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  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