Realtime Compute for Apache Flink:MySQL

背景情報

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

特徴

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

前提条件

MySQL CDC ソーステーブルを使用する前に、MySQL の設定に記載されている通り、MySQL データベースを設定する必要があります。以下の設定が必要です。

制限事項

注意事項

• 各 MySQL CDC データソースに対して、異なる Server ID を明示的に設定します

  Server ID の目的

  各 MySQL CDC データソースに対して、異なる Server ID を明示的に設定する必要があります。複数の MySQL CDC データソースが同じ Server ID を共有し、接続を再利用できない場合、binlog オフセットが乱れ、過剰読み取りまたは不足読み取りが発生します。

  さまざまなシナリオにおける Server ID の設定

  DDL 文で Server ID を指定できますが、DDL オプションではなく動的ヒントで設定することを推奨します。

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

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

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

    ## Server ID の範囲を指定します。範囲内の利用可能な Server 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）またはスロークエリログを確認し、実行された実際の SELECT ステートメントを確認してください。

例

• 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 ソース
    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 シンク
    print.enabled: true
    sink.print.logger: true

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

• 仕組み

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

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

• メタデータ

  メタデータは、シャーディングされたデータベースおよびテーブルのマージシナリオで非常に有用です。マージ後も、ビジネスでは各行のデータのソースデータベースおよびテーブルを識別する必要がある場合がよくあります。メタデータカラムを使用すると、この情報をアクセスできます。そのため、メタデータカラムを使用して、複数のシャーディングされたテーブルを簡単に 1 つの宛先テーブルにマージできます。

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

  メタデータキー	メタデータ型	説明
  database_name	STRING NOT NULL	行を含むデータベースの名前。
  table_name	STRING NOT NULL	行を含むテーブルの名前。
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	データベース内で行が変更された時刻。レコードが binlog ではなくテーブルの既存の履歴データに由来する場合、この値は常に 0 になります。
  op_type	STRING NOT NULL	行の変更タイプ。 +I：INSERT メッセージ -D：DELETE メッセージ -U：UPDATE_BEFORE メッセージ +U：UPDATE_AFTER メッセージ 説明 Ververica Runtime（VVR）8.0.7 以降でのみサポートされます。
  query_log	STRING NOT NULL	この行に対応する MySQL クエリログレコードを読み取ります。 説明 MySQL は、クエリログを記録するために binlog_rows_query_log_events パラメーターを有効化している必要があります。

  次の例は、MySQL インスタンス内の異なるシャーディングされたデータベースの複数の orders テーブルをマージし、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 コネクタは、完全データの同時読み取りをサポートしており、これによりデータ読み込み効率が向上します。Realtime Compute for Apache Flink コンソールの Autopilot と組み合わせることで、同時読み取りが完了した後の増分フェーズ中に自動的にスケールインし、コンピューティングリソースを節約します。

  Realtime Compute 開発コンソールでは、リソース構成ページで基本モードまたはエキスパートモードのいずれかでジョブの並列処理の次数を設定できます。違いは次のとおりです：

  • 基本モードで設定された並列処理の次数は、ジョブ全体のグローバルな並列処理の次数です。

  • エキスパートモードでは、必要に応じて特定の VERTEX の並列処理の次数を設定できます。

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

  重要

  基本モードまたはエキスパートモードのいずれを使用する場合でも、テーブルで宣言された server-id 範囲は、ジョブの並列処理の次数以上である必要があります。たとえば、server-id 範囲が 5404-5412 の場合、8 個の一意な server ID があります。したがって、ジョブは最大 8 個の並列タスクを持つことができます。同じ MySQL インスタンスに対する異なるジョブは、重複しない server-id 範囲を持つ必要があります。各ジョブは、明示的に異なる server-id を設定する必要があります。

• Autopilot 自動スケールイン

  完全データフェーズでは、大量の履歴データが蓄積されます。読み取り効率を向上させるために、通常、履歴データは同時読み取りされます。一方、増分 binlog フェーズでは、binlog データ量が少なく、グローバルな順序付けを維持する必要があるため、通常は単一の並列処理で十分です。Autopilot は、完全フェーズと増分フェーズの間でこれらの異なる要件を満たすために、パフォーマンスとリソースのバランスを自動的に調整します。

  Autopilot は、MySQL CDC ソースの各タスクのトラフィックを監視します。binlog フェーズに移行した際に、1 つのタスクのみが binlog 読み取りを処理し、他のタスクがアイドル状態になる場合、Autopilot は自動的にソースの CU 数および並列処理の次数を削減します。Autopilot を有効化するには、ジョブの運用保守ページで Autopilot モードを Active に設定します。

  説明

  並列処理の次数をスケールダウンするためのデフォルトの最小トリガー間隔は 24 時間です。Autopilot パラメーターおよび詳細については、「Autopilot の設定」をご参照ください。

• 起動モード

  scan.startup.mode オプションを使用して、MySQL CDC ソーステーブルの起動モードを指定できます。有効な値は次のとおりです：

  • initial（デフォルト）：最初の起動時にデータベーステーブルの完全読み取りを実行し、その後増分モードに切り替えて binlog を読み取ります。

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

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

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

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

  例：

  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 モードの binlog ファイル名を指定します。
      'scan.startup.specific-offset.pos' = '4', -- specific-offset モードの binlog 位置を指定します。
      '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 回限りのセマンティクスが保証されます。

    • 指定されたカラムが更新される場合、最低限 1 回のセマンティクスのみが保証されます。ただし、下流システムと組み合わせて、下流のプライマリキーを指定し、べき等操作を使用することで、データの正確性を確保できます。

• RDS MySQL からのバックアップログの読み取り

  MySQL CDC ソーステーブルは、Alibaba Cloud RDS MySQL からのバックアップログの読み取りをサポートしています。この機能は、完全スナップショットフェーズに長い時間がかかる場合に特に有用です。この場合、ローカルの binlog ファイルは自動的にクリーンアップされる可能性がありますが、手動または自動でアップロードされたバックアップファイルは依然として存在します。

  例：

  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 インスタンスから読み取る場合、この方法はデータベースへの負荷を増加させます。詳細については、「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 では、オペレーターチェーンを無効化するとシリアル化の問題が発生します。

binlog 読み取りの高速化

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

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

  • 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：1 回の反復で処理される最大イベント数。デフォルト値は 2048 です。

  • debezium.poll.interval.ms：コネクタが新しい変更イベントを要求する前に待機するミリ秒数。デフォルト値は 1000 ms（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 ソース
  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 のエンタープライズ版は、オープンソースコミュニティ版の約 2 倍である 85 MB/s の速度で binlog を消費します。binlog 生成速度が 85 MB/s（512 MB のファイルを 6 秒ごとに生成する速度に相当）を超える場合、Flink ジョブの遅延が継続的に増加します。binlog 生成速度が低下すると、遅延は徐々に減少します。binlog ファイルに大規模なトランザクションが含まれている場合、処理遅延が一時的に増加し、トランザクションのログの読み取りが完了すると減少します。

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") // キャプチャ対象のデータベースを設定します。
        .tableList("yourDatabaseName.yourTableName") // キャプチャ対象のテーブルを設定します。
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // SourceRecord を JSON 文字列に変換します。
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // チェックポイントを有効化します。
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL ソース")
      // 並列ソースタスクを 4 つに設定します。
      .setParallelism(4)
      .print().setParallelism(1); // メッセージの順序を維持するために、シンクの並列処理の次数を 1 に設定します。
    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 の問題」をご参照ください。