Realtime Compute for Apache Flink:MySQL

背景情報

MySQL コネクタは、MySQL プロトコルと互換性のあるすべてのデータベース（ApsaraDB RDS for MySQL、PolarDB for MySQL、OceanBase（MySQL モード）、および自己管理 MySQL データベース）をサポートしています。

主な機能

MySQL CDC ソーステーブルは、まずデータベースから履歴の完全データを読み取り、その後シームレスにバイナリログの読み取りに切り替えます。これにより、データの欠落や重複が発生しません。障害が発生した場合でも、データ処理は 1 回限りのセマンティクスを保証します。MySQL CDC ソーステーブルは、完全データの並列読み取りをサポートし、ロックレス操作と再開可能な転送を実現するための増分スナップショットアルゴリズムを使用します。詳細については、「MySQL CDC ソーステーブルについて」をご参照ください。

前提条件

MySQL CDC ソーステーブルを使用する前に、「MySQL の設定」の手順を完了してください。これらの手順により、MySQL CDC ソーステーブルの使用に必要な前提条件が満たされます。

使用制限

注意事項

• 各 MySQL CDC データソースには、一意の server ID が必要です。

  server ID の目的

  各 MySQL CDC データソースには、一意の server ID が必要です。複数のデータソースが同じ server ID を共有し、接続を再利用できない場合、バイナリログオフセットが不整合になる可能性があり、データの欠落または重複が発生します。

  さまざまなシナリオにおける server ID の構成

  DDL 文で server ID を指定できますが、DDL パラメーターではなく動的ヒントを使用することを推奨します。

  • 並列処理の次数 = 1 または増分スナップショットが無効化されている場合

    ## 増分スナップショットが無効化されている場合、または並列処理の次数が 1 の場合に特定の server ID を指定します。
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

  • 並列処理の次数 > 1 かつ増分スナップショットが有効化されている場合

    ## server ID の範囲を指定します。範囲内の利用可能な ID 数が、並列処理の次数以上であることを確認してください。たとえば、並列処理の次数が 3 の場合：
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

  • CTAS を使用したデータ同期

    CTAS を使用してデータを同期する場合、CDC データソースの構成が同一であれば、それらは自動的に再利用されます。この場合、複数の CDC データソースに同じ server ID を割り当てることができます。詳細については、「例 4：複数の CTAS ステートメント」をご参照ください。

  • 再利用できない複数の非 CTAS ソーステーブル

    ジョブに複数の MySQL CDC ソーステーブルが含まれており、CTAS を使用していない場合、データソースは再利用できません。各 CDC ソーステーブルに一意の server ID を割り当ててください。同様に、増分スナップショットが有効化されており、並列処理の次数が 1 より大きい場合は、server ID の範囲を指定してください。

    select * from 
      source_table1 /*+ OPTIONS('server-id'='123456-123457') */
    left join 
      source_table2 /*+ OPTIONS('server-id'='123458-123459') */
    on source_table1.id=source_table2.id;

• シンクテーブル

  • 自動インクリメント主キーは DDL で宣言されません。MySQL がデータ書き込み時に自動的に埋めます。

  • 少なくとも 1 つの非主キーフィールドを宣言する必要があります。そうでない場合、エラーが発生します。

  • DDL 内の NOT ENFORCED は、Flink が主キー検証を強制しないことを意味します。主キーの正確性および整合性を保証する必要があります。詳細については、「妥当性チェック」をご参照ください。

• ディメンションテーブル

  インデックスを使用したクエリの高速化のため、JOIN 条件のフィールドはインデックス定義の順序（左端プレフィックスルール）と一致している必要があります。たとえば、インデックスが (a, b, c) の場合、JOIN 条件は ON t.a = x AND t.b = y である必要があります。

  Flink が生成する SQL は、オプティマイザーによって書き換えられる可能性があり、インデックスが使用されないことがあります。インデックスが実際に使用されているかどうかを確認するには、MySQL の実行計画（EXPLAIN）またはスロークエリログを確認してください。

使用例

• CDC ソーステーブル

  CREATE TEMPORARY TABLE mysqlcdc_source (
     order_id INT,
     order_date TIMESTAMP(0),
     customer_name STRING,
     price DECIMAL(10, 5),
     product_id INT,
     order_status BOOLEAN,
     PRIMARY KEY(order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    order_id INT,
    customer_name STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT order_id, customer_name FROM mysqlcdc_source;

• ディメンションテーブル

  CREATE TEMPORARY TABLE datagen_source(
    a INT,
    b BIGINT,
    c STRING,
    `proctime` AS PROCTIME()
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_dim (
    a INT,
    b VARCHAR,
    c VARCHAR
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    a INT,
    b STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT T.a, H.b
  FROM datagen_source AS T JOIN mysql_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.a = H.a;

• 結果テーブル

  CREATE TEMPORARY TABLE datagen_source (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_sink (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  INSERT INTO mysql_sink
  SELECT * FROM datagen_source;

• データインジェストソース

  source:
    type: mysql
    name: MySQL Source
    hostname: ${mysql.hostname}
    port: ${mysql.port}
    username: ${mysql.username}
    password: ${mysql.password}
    tables: ${mysql.source.table}
    server-id: 7601-7604
  
  sink:
    type: values
    name: Values Sink
    print.enabled: true
    sink.print.logger: true

MySQL CDC ソーステーブルについて

• 実装原則

  MySQL CDC ソーステーブルが起動すると、テーブル全体をスキャンし、主キーに基づいて複数のチャンクに分割し、バイナリログオフセットを記録します。その後、増分スナップショットアルゴリズムを使用して SELECT 文で各チャンクからデータを順次読み取ります。ジョブは定期的にチェックポイントを実行し、完了したチャンクを記録します。フェールオーバーが発生した場合、未完了のチャンクから読み取りを再開します。すべてのチャンクが読み取られた後、以前に記録されたバイナリログオフセットから増分変更レコードを読み取ります。Flink ジョブは定期的にチェックポイントを実行し続け、バイナリログオフセットを記録することで、1 回限りのセマンティクスを保証します。

  増分スナップショットアルゴリズムの詳細については、「MySQL CDC コネクタ」をご参照ください。

• メタデータ

  メタデータは、複数のデータベースおよびテーブルからデータをマージおよび同期するシャーディングシナリオで役立ちます。このような場合、各レコードのソースデータベースおよびテーブルを区別する必要があることがよくあります。メタデータカラムは、ソーステーブルのデータベース名およびテーブル名情報へのアクセスを提供し、複数のシャーディングされたテーブルを単一の送信先テーブルに簡単にマージできるようにします。

  MySQL CDC ソースはメタデータカラム構文をサポートしており、以下のメタデータにアクセスできます。

  メタデータキー	メタデータ型	説明
  database_name	STRING NOT NULL	レコードを含むデータベース名。
  table_name	STRING NOT NULL	レコードを含むテーブル名。
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	データベース内のレコードの変更時間。レコードがバイナリログではなく履歴データからのものである場合、この値は常に 0 です。
  op_type	STRING NOT NULL	レコードの変更タイプ。 +I: INSERT メッセージ -D: DELETE メッセージ -U: UPDATE_BEFORE メッセージ +U: UPDATE_AFTER メッセージ 説明 VVR 8.0.7 以降でのみサポートされます。
  query_log	STRING NOT NULL	行に対応する MySQL クエリログレコード。 説明 MySQL で binlog_rows_query_log_events パラメーターが有効化されている必要があります。

  以下のコード例は、MySQL インスタンス内の異なるデータベースからの複数のシャーディングされた注文テーブルを、Hologres の単一の holo_orders テーブルにマージします。

  CREATE TEMPORARY TABLE mysql_orders (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,  -- データベース名を読み取ります。
    table_name STRING METADATA  FROM 'table_name' VIRTUAL, -- テーブル名を読み取ります。
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL, -- 変更時間を読み取ります。
    op_type STRING METADATA FROM 'op_type' VIRTUAL, -- 変更タイプを読み取ります。
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database-name' = 'mydb_.*', -- 複数のシャーディングされたデータベースを正規表現でマッチさせます。
    'table-name' = 'orders_.*'   -- 複数のシャーディングされたテーブルを正規表現でマッチさせます。
  );
  
  INSERT INTO holo_orders SELECT * FROM mysql_orders;

  上記のコードに基づき、WITH 句で scan.read-changelog-as-append-only.enabled パラメーターを true に設定した場合、結果テーブルの主キー構成によって出力が異なります。

  • 結果テーブルの主キーが order_id の場合、出力にはソーステーブルからの各主キーの最後の変更のみが含まれます。最後の変更が削除操作であった主キーの場合、結果テーブルには同じ主キーと op_type が -D のレコードが表示されます。

  • 結果テーブルの主キーが order_id、operation_ts、および op_type の場合、出力にはソーステーブルからの各主キーの完全な変更履歴が含まれます。

• 正規表現のサポート

  MySQL CDC ソーステーブルは、テーブル名またはデータベース名で正規表現を使用して、複数のテーブルまたはデータベースをマッチさせることをサポートしています。以下のコード例は、正規表現を使用して複数のテーブルを指定する方法を示しています。

  CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})', -- 複数のデータベースを正規表現でマッチさせます。
    'table-name' = '(t[5-8]|tt)' -- 複数のテーブルを正規表現でマッチさせます。
  );

  例の正規表現の説明：

  • ^(test).* はプレフィックスマッチングの例です。この式は、test1 や test2 のように "test" で始まるデータベース名にマッチします。

  • .*[p$] はサフィックスマッチングの例です。この式は、cdcp や edcp のように "p" で終わるデータベース名にマッチします。

  • txc は完全一致です。これはデータベース名 "txc" にマッチします。

  完全パスのテーブル名をマッチさせる場合、MySQL CDC はデータベース名とテーブル名、つまり database-name.table-name をマッチングパターンとして使用してテーブルを一意に識別します。たとえば、パターン (^(test).*|^(tpc).*|txc|.*[p$]|t{2}).(t[5-8]|tt) は、データベース内のテーブル txc.tt および test2.test5 にマッチできます。

  重要

  SQL ジョブ構成では、table-name および database-name は、複数のテーブルまたはデータベースを指定するためのカンマ区切りのリストをサポートしていません。

  • 複数のテーブルにマッチさせる場合、または複数の正規表現を使用する場合は、縦棒（|）で連結し、括弧で囲んでください。たとえば、テーブル 'user' および 'product' を読み取るには、table-name を (user|product) に設定します。

  • 正規表現にカンマが含まれる場合は、縦棒（|）演算子を使用して書き換えてください。たとえば、mytable_\d{1, 2} を同等の (mytable_\d{1}|mytable_\d{2}) に書き換えて、カンマの使用を避けてください。

• 同時実行制御

  MySQL コネクタは、完全データの同時読み取りをサポートし、データ読み込み効率を向上させます。Flink Realtime Compute コンソールの Autopilot 自動チューニング機能と組み合わせることで、同時完全データ読み取り後に増分フェーズで自動的にスケールインし、計算リソースを節約します。

  Realtime Compute 開発コンソールでは、基本モードまたはエキスパートモードでジョブの並列処理の次数を設定できます。同時実行数の設定の違いは以下のとおりです。

  • 基本モードで設定された同時実行数は、ジョブ全体に適用されます。

  • エキスパートモードでは、特定の頂点の並列処理の次数を設定できます。

  リソース構成の詳細については、「ジョブデプロイメント情報の設定」をご参照ください。

  重要

  モードに関係なく、同時実行数を設定する場合、テーブルで宣言された server-id の範囲は、ジョブの並列処理の次数以上である必要があります。たとえば、server-id の範囲が 5404-5412 の場合、8 つの一意の server ID が存在するため、ジョブは最大 8 つの同時タスクを持つことができます。同じ MySQL インスタンスにアクセスする異なるジョブは、重複しない server-id の範囲を持つ必要があり、各ジョブは一意の server-id を明示的に構成する必要があります。

• Autopilot 自動スケーリング

  完全データフェーズでは、大量の履歴データが蓄積されます。読み取り効率を向上させるために、通常は同時読み取りが使用されます。バイナリログ増分フェーズでは、バイナリログデータ量が少なく、グローバルな順序を保証するため、通常はシングルスレッド読み取りで十分です。完全フェーズと増分フェーズで異なるリソース要件は、自動チューニング機能によって自動的にバランスされます。

  自動チューニングは、MySQL CDC ソースの各タスクのトラフィックを監視します。ジョブがバイナリログフェーズに入ると、1 つのタスクのみがバイナリログ読み取りを担当し、他のタスクがアイドル状態の場合、自動チューニングはソースの CU 数と同時実行数を自動的に削減します。自動チューニングを有効にするには、ジョブ操作ページで自動チューニングモードを Active に設定します。

  説明

  同時実行数を削減するための最小トリガー間隔は、デフォルトで 24 時間です。自動チューニングパラメーターおよび詳細については、「自動チューニングの設定」をご参照ください。

• 起動モード

  scan.startup.mode 構成項目を使用して、MySQL CDC ソーステーブルの起動モードを指定します。有効な値：

  • initial（デフォルト）：初回起動時に、データベーステーブルの完全読み取りを実行し、その後増分バイナリログ読み取りに切り替えます。

  • earliest-offset：スナップショットフェーズをスキップし、利用可能な最も古いバイナリログオフセットから読み取りを開始します。

  • latest-offset：スナップショットフェーズをスキップし、バイナリログの末尾から読み取りを開始します。このモードでは、ソーステーブルはジョブ開始後に発生したデータ変更のみを読み取ります。

  • specific-offset：スナップショットフェーズをスキップし、指定されたバイナリログオフセットから読み取りを開始します。オフセットは、バイナリログファイル名と位置、または GTID セットで指定できます。

  • timestamp：スナップショットフェーズをスキップし、指定されたタイムスタンプからバイナリログイベントの読み取りを開始します。

  使用例：

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'scan.startup.mode' = 'earliest-offset', -- 最も古いオフセットから開始します。
      'scan.startup.mode' = 'latest-offset', -- 最新のオフセットから開始します。
      'scan.startup.mode' = 'specific-offset', -- 特定のオフセットから開始します。
      'scan.startup.mode' = 'timestamp', -- 特定のタイムスタンプから開始します。
      'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- specific-offset モードのバイナリログファイル名を指定します。
      'scan.startup.specific-offset.pos' = '4', -- specific-offset モードのバイナリログ位置を指定します。
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- specific-offset モードの GTID セットを指定します。
      'scan.startup.timestamp-millis' = '1667232000000' -- timestamp モードの起動タイムスタンプを指定します。
      ...
  )

  重要

  • MySQL ソースは、チェックポイント時に INFO レベルのログで現在のオフセットを出力します。ログプレフィックスは Binlog offset on checkpoint {checkpoint-id} です。このログは、特定のチェックポイントオフセットからジョブを開始するのに役立ちます。

  • 読み取られるテーブルがスキーマ変更を受けている場合、earliest-offset、specific-offset、または timestamp モードから開始するとエラーが発生する可能性があります。これは、Debezium リーダーが内部的に最新のテーブルスキーマを維持しており、スキーマが一致しない以前のデータが正しく解析できないためです。

• キーレス CDC ソーステーブルについて

  • キーレステーブルを使用するには、scan.incremental.snapshot.chunk.key-column を設定する必要があり、選択されたカラムは NULL 不可である必要があります。

  • キーレス CDC ソーステーブルの処理セマンティクスは、scan.incremental.snapshot.chunk.key-column で指定されたカラムの動作によって決定されます。

    • 指定されたカラムが更新されない場合、1 回限りのセマンティクスが保証されます。

    • 指定されたカラムが更新される場合、at-least-once セマンティクスのみが保証されます。ただし、下流システムで主キーを指定し、べき等操作を使用することでデータの正確性を保証できます。

• Alibaba Cloud ApsaraDB RDS for MySQL バックアップログの読み取り

  MySQL CDC ソーステーブルは、Alibaba Cloud ApsaraDB RDS for MySQL からバックアップログを読み取ることをサポートしています。これは、完全データフェーズに時間がかかり、ローカルのバイナリログファイルが自動的にクリーンアップされるが、自動または手動でアップロードされたバックアップファイルがまだ存在する場合に役立ちます。

  使用例：

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'rds.region-id' = 'cn-beijing',
      'rds.access-key-id' = 'xxxxxxxxx', 
      'rds.access-key-secret' = 'xxxxxxxxx', 
      'rds.db-instance-id' = 'rm-xxxxxxxxxxxxxxxxx', 
      'rds.main-db-id' = '12345678',
      'rds.download.timeout' = '60s'
      ...
  )

• CDC ソースの再利用の有効化

  単一のジョブで複数の MySQL CDC ソーステーブルが複数の Binlog クライアントを起動すると、すべてのソーステーブルが同じインスタンスにある場合、データベースへの負荷が増加します。詳細については、「MySQL CDC よくある質問」をご参照ください。

  ソリューション

  VVR 8.0.7 以降は MySQL CDC ソースの再利用をサポートしています。再利用により、結合可能な MySQL CDC ソーステーブルがマージされます。マージは、データベース名、テーブル名、および server-id を除いて、ソーステーブルの構成が同一である場合に発生します。エンジンは、同じジョブ内の MySQL CDC ソースを自動的にマージします。

  手順

  1. SQL ジョブで SET コマンドを使用します。

     SET 'table.optimizer.source-merge.enabled' = 'true';
     
     # (VVR 8.0.8 および 8.0.9) さらに以下を設定します。
     SET 'sql-gateway.exec-plan.enabled' = 'false';

     VVR 11.1 以降では、再利用がデフォルトで有効化されています。

  2. ジョブをステートレスで起動します。ソースの再利用構成を変更すると、ジョブのトポロジーが変更されます。ジョブはステートレスで起動する必要があります。そうしないと、ジョブの起動に失敗したり、データが失われたりする可能性があります。ソースがマージされると、MergetableSourceScan ノードが表示されます。

  重要

  • 再利用を有効化した後、演算子チェーンを無効にしないことを推奨します。pipeline.operator-chaining を false に設定すると、データシリアル化および逆シリアル化のオーバーヘッドが増加します。マージされるソースが多いほど、オーバーヘッドは大きくなります。

  • VVR 8.0.7 では、演算子チェーンを無効にするとシリアル化の問題が発生します。

バイナリログ読み取りの高速化

MySQL コネクタがソーステーブルまたはデータインジェストソースとして使用される場合、増分フェーズ中にバイナリログファイルを解析してさまざまな変更メッセージを生成します。バイナリログファイルは、すべてのテーブル変更をバイナリ形式で記録します。以下の方法でバイナリログファイルの解析を高速化します。

• 解析フィルター構成の有効化

  • 構成項目 scan.only.deserialize.captured.tables.changelog.enabled を使用します。指定されたテーブルの変更イベントのみを解析します。

• Debezium パラメーターの最適化

  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50

  • debezium.max.queue.size：ブロッキングキューが保持できるレコードの最大数。Debezium がデータベースからイベントストリームを読み取る際、イベントを下流に書き込む前にブロッキングキューに配置します。デフォルト値は 8192 です。

  • debezium.max.batch.size：コネクタが各反復で処理するイベントの最大数。デフォルト値は 2048 です。

  • debezium.poll.interval.ms：コネクタが新しい変更イベントを要求する前に待機するミリ秒数。デフォルト値は 1000 ミリ秒（1 秒）です。

使用例：

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium 構成
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- 解析フィルターを有効化
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- 指定されたテーブルの変更イベントのみを解析します。
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium 構成
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # 解析フィルターを有効化
  scan.only.deserialize.captured.tables.changelog.enabled: true

MySQL CDC のエンタープライズ版は、バイナリログ消費能力が 85 MB/秒であり、これはオープンソースコミュニティ版の約 2 倍です。バイナリログ生成レートが 85 MB/秒（つまり、6 秒ごとに 512 MB のファイル）を超えると、Flink ジョブのレイテンシーは継続的に増加します。バイナリログ生成レートが低下すると、レイテンシーは徐々に減少します。バイナリログファイルに大規模なトランザクションが含まれている場合、処理レイテンシーは一時的に増加し、トランザクションのログが処理された後に減少する可能性があります。

MySQL CDC DataStream API

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // set captured database
        .tableList("yourDatabaseName.yourTableName") // set captured table
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // enable checkpoint
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
    env.execute("Print MySQL Snapshot + Binlog");
  }
}

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>

よくある質問

CDC ソーステーブルの使用時に発生する一般的な問題については、「CDC の問題」をご参照ください。