Realtime Compute for Apache Flink:MySQL

背景情報

MySQL コネクタは、ApsaraDB RDS for MySQL、PolarDB for MySQL、OceanBase (MySQL モード)、セルフマネージド MySQL など、MySQL プロトコルと互換性のあるすべてのデータベースをサポートします。

特徴

MySQL 変更データキャプチャ (CDC) ソーステーブルは、最初にデータベースから完全な既存データを読み取るストリーミングソーステーブルです。その後、スムーズにバイナリログ (Binlog) の読み取りに切り替えることで、データの欠落や重複がないことを保証します。障害が発生した場合でも、Exactly-Once セマンティクスが保証されます。MySQL CDC ソーステーブルは、完全データの同時読み取りをサポートし、増分スナップショットアルゴリズムを使用して、ロックフリーの読み取りと再開可能なデータ転送を実装します。詳細については、「MySQL CDC ソーステーブルについて」をご参照ください。

前提条件

MySQL CDC ソーステーブルを使用する前に、「MySQL の構成」で説明されているように MySQL を構成して、前提条件を満たす必要があります。

制限事項

注意

• 各 MySQL CDC データソースには、明示的に異なるサーバー ID を構成する必要があります

  サーバー ID の目的

  複数の MySQL CDC データソースが同じサーバー ID を共有し、再利用できない場合、バイナリログのオフセットが乱れる可能性があります。これにより、データが複数回読み取られたり、欠落したりする可能性があります。

  異なるシナリオでのサーバー ID の構成

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

  • 並列度 = 1 または増分スナップショットが無効

    ## 増分スナップショットフレームワークが無効になっているか、並列度が 1 の場合、特定のサーバー ID を指定できます。
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

  • 並列度 > 1 かつ増分スナップショットが有効

    ## サーバー ID の範囲を指定する必要があります。範囲内の利用可能なサーバー ID の数は、並列度以上でなければなりません。並列度が 3 であると仮定します。
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

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

    CTAS を使用してデータ同期を行う場合、CDC データソースの構成が同じであれば、自動的に再利用されます。この場合、複数の CDC データソースに同じサーバー ID を構成できます。詳細については、「例 4：複数の CTAS 文」をご参照ください。

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

    ジョブに複数の MySQL CDC ソーステーブルが含まれており、同期に CTAS 文を使用しない場合、データソースは再利用できません。各 CDC ソーステーブルに異なるサーバー ID を提供する必要があります。同様に、増分スナップショットフレームワークが有効で、並列度が 1 より大きい場合は、サーバー 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 はオプティマイザーによって再書き込みされる可能性があり、これにより実際のデータベースクエリ中にインデックスが使用されなくなることがあります。インデックスが使用されているかどうかを確認するには、実行計画 (EXPLAIN) または MySQL の Redis スローログをチェックして、実際に実行される 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 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 ジョブは定期的なチェックポイントを実行し続け、バイナリログオフセットを記録します。ジョブがフェールオーバーした場合、最後に記録されたバイナリログオフセットから処理を再開します。このプロセスにより、Exactly-Once セマンティクスが実現されます。

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

• メタデータ

  メタデータは、シャーディングされたデータベースとテーブルをマージして同期するシナリオで役立ちます。これは、マージ後、ビジネスでは各データのソースデータベースとテーブルを区別したいことが多いためです。メタデータ列を使用して、ソーステーブルのデータベースとテーブル名情報にアクセスできます。したがって、メタデータ列を使用して、複数のシャーディングされたテーブルを単一の宛先テーブルに簡単にマージできます。

  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 メッセージ 説明 Realtime Compute for Apache Flink 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 for Apache Flink 開発コンソールでは、リソース構成ページで基本モードまたはエキスパートモードでジョブの同時実行数を設定できます。違いは次のとおりです：

  • 基本モードで設定された同時実行数は、ジョブ全体のグローバルな同時実行数です。

  • エキスパートモードは、必要に応じて特定の VERTEX の同時実行数を設定することをサポートしています。

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

  重要

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

• Autopilot 自動スケールイン

  完全データフェーズでは、大量の既存データが蓄積されます。読み取り効率を向上させるために、既存データは通常、同時に読み取られます。増分バイナリログフェーズでは、バイナリログデータの量が少なく、グローバルな順序を保証する必要があるため、通常は単一の同時読み取りで十分です。完全フェーズと増分フェーズの異なるリソース要件は、自動チューニング機能によって自動的にバランスが取られ、高性能とリソース効率の両方を実現できます。

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

  説明

  並列処理の次数を減らすためのデフォルトの最小トリガー間隔は 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 モードで 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 を設定する必要があります。非ヌルフィールドのみを選択できます。

  • プライマリキーのない CDC ソーステーブルの処理セマンティクスは、scan.incremental.snapshot.chunk.key-column で指定された列の動作によって決まります：

    • 指定された列が更新されない場合、Exactly-Once セマンティクスが保証されます。

    • 指定された列が更新される場合、At-Least-Once セマンティクスのみが保証されます。ただし、プライマリキーを持ち、べき等操作を使用する下流のシンクと列を組み合わせることで、データの正確性を保証できます。

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

  MySQL CDC ソーステーブルは、Alibaba Cloud 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 よくある質問」をご参照ください。

  ソリューション

  Realtime Compute for Apache Flink 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 コネクタをソーステーブルまたはデータインジェストのデータソースとして使用する場合、増分フェーズ中にバイナリログファイルを解析してさまざまな変更メッセージを生成します。バイナリログファイルは、すべてのテーブルの変更をバイナリ形式で記録します。次の方法でバイナリログファイルの解析を高速化できます。

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

  • 構成項目 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 Enterprise Edition のバイナリログ消費能力は 85 MB/s で、オープンソースコミュニティバージョンの約 2 倍です。バイナリログファイルの生成速度が 85 MB/s を超える場合 (つまり、512 MB のファイルが 6 秒ごとに 1 つ生成される場合)、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") // キャプチャするデータベースを設定
        .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 Source")
      // 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 の問題」をご参照ください。