背景信息

MySQL連接器支持所有兼容MySQL協議的數據庫，包括RDS MySQL、PolarDB for MySQL、OceanBase（MySQL模式）或者自建MySQL。

特色功能

MySQL的CDC源表，即MySQL的流式源表，會先讀取數據庫的歷史全量數據，并平滑切換到Binlog讀取上，保證不多讀一條也不少讀一條數據。即使發生故障，也能保證通過Exactly Once語義處理數據。MySQL CDC源表支持并發地讀取全量數據，通過增量快照算法實現了全程無鎖和斷點續傳，詳情可參見關于MySQL CDC源表。

前提條件

• CDC源表和數據攝入數據源

  在使用MySQL CDC源表前，必須先按照配置MySQL進行操作，這些操作主要為了滿足使用MySQL CDC源表的前提條件：

  • MySQL和VVP的網絡連通。

  • MySQL服務器配置要求：

    • MySQL版本為5.6，5.7或8.0.x。

    • 已開啟了Binlog。

    • Binlog格式已設置為ROW。

    • binlog_row_image已設置為FULL。

  • 已在MySQL配置文件中配置了交互超時或等待超時參數。

  • 已創建MySQL用戶，并授予了SELECT、SHOW DATABASES、REPLICATION SLAVE和REPLICATION CLIENT權限。

• 維表和結果表

  • 已創建MySQL數據庫和表，詳情請參見RDS MySQL創建數據庫和賬號、PolarDB MySQL創建數據庫和賬號或自建MySQL創建數據庫和賬號。

  • 已設置IP白名單，詳情請參見RDS MySQL白名單設置、PolarDB MySQL白名單設置或自建MySQL白名單設置。

使用限制

• CDC源表和數據攝入數據源

  • 僅VVR 4.0.8及以上引擎版本支持無鎖讀取和并發讀取功能。

  • 請根據MySQL版本選擇合適的引擎版本，MySQL版本支持情況如下表所示。

    您可以通過執行select version()命令來查看MySQL的版本。

    VVR版本	支持的MySQL版本
    VVR 4.0.8 ～ VVR 4.0.10	5.7
    	8.0.x
    VVR 4.0.11及以上版本	5.6.x
    	5.7.x
    	8.0.x

    重要

    為了確保RDS MySQL 5.6.x版本的正常運行，默認已開啟增量快照功能（即scan.incremental.snapshot.enabled=true），且不支持關閉增量快照功能，而RDS MySQL 6.0.8和8.0.1版本的數據庫已解除該限制，即支持關閉增量快照功能。建議您不要關閉增量快照功能，因為關閉增量快照功能會鎖定MySQL數據庫，可能會對線上業務處理性能產生影響。

  • MySQL CDC源表暫不支持定義Watermark。如果您需要進行窗口聚合，您可以采用非窗口聚合的方式，詳情請參見不支持定義Watermark，那如何進行窗口聚合？。

  • MySQL的CDC源表需要一個有特定權限（包括SELECT、SHOW DATABASES、REPLICATION SLAVE和REPLICATION CLIENT）的MySQL用戶，才能讀取全量和增量數據。

  • 當結合CTAS和CDAS整庫同步語法使用時，MySQL CDC源表可以同步部分Schema變更，支持的變更類型詳情請參見表結構變更同步策略。在其他使用場景下，MySQL CDC源表無法同步Schema變更操作。

  • MySQL CDC源表無法同步Truncate操作。

  • 對于RDS MySQL，不建議通過備庫或只讀從庫讀取數據。因為RDS MySQL的備庫和只讀從庫Binlog保留時間默認很短，可能由于Binlog過期清理，導致作業無法消費Binlog數據而報錯。

  • MySQL CDC源表不支持讀取PolarDB MySQL版1.0.19及以前版本的多主架構集群（什么是多主集群？）。PolarDB MySQL版1.0.19及以前版本的多主架構集群產生的Binlog可能出現重復Table id，導致CDC源表Schema映射錯誤，從而解析Binlog數據報錯。PolarDB MySQL版在高于1.0.19的版本進行適配，保證Binlog內Table id不會出現重復，從而避免解析報錯。

• 維表和結果表

  • Flink計算引擎VVR 4.0.11及以上版本支持MySQL連接器。

  • 語義上可以保證At-Least-Once，在結果表有主鍵的情況下，冪等可以保證數據的正確性。

注意事項

• CDC源表和數據攝入數據源

  • 每個MySQL CDC數據源需顯式配置不同的Server ID。

    • Server ID作用

      每個同步數據庫數據的客戶端，都會有一個唯一ID，即Server ID。MySQL SERVER會根據該ID來維護網絡連接以及Binlog位點。因此如果有大量不同的Server ID的客戶端一起連接MySQL SERVER，可能導致MySQL SERVER的CPU陡增，影響線上業務穩定性。

      此外，如果多個MySQL CDC數據源共享相同的Server ID，且數據源之間無法復用時，會導致Binlog位點錯亂，多讀或少讀數據。還可能出現Server ID沖突的報錯，詳情請參見上下游存儲。因此建議每個MySQL CDC數據源都配置不同的Server ID。

    • Server ID配置方式

      Server ID可以在DDL中指定，也可以通過動態Hints配置。

      建議通過動態Hints來配置Server ID，而不是在DDL參數中配置Server ID。動態Hints詳情請參見動態Hints。

    • 不同場景下Server ID的配置

      • 未開啟增量快照框架或并行度為1

        當未開啟增量快照框架或并行度為1時，可以指定一個特定的Server ID。

        SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

      • 開啟增量快照框架且并行度大于1

        當開啟增量快照框架且并行度大于1時，需要指定Server ID范圍，要保證范圍內可用的Server ID數量不小于并行度。假設并行度為3，可以如下配置：

        SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

      • 結合CTAS進行數據同步

        當結合CTAS進行數據同步時，如果CDC數據源配置相同，會自動對數據源進行復用，此時可以為多個CDC數據源配置相同的Server ID。詳情請參見代碼示例四：多CTAS語句。

      • 同一作業包含多個MySQL CDC源表（非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;

  • 僅VVR 4.0.8及以上版本支持全量階段的無鎖讀取、并發讀取、斷點續傳等功能。

    如果您使用的是VVR 4.0.8以下版本，需要對MySQL用戶授予RELOAD權限用來獲取全局讀鎖，保證數據讀取的一致性。全局讀鎖會阻塞寫入操作，持鎖時間可能達到秒級，因此可能對線上業務造成影響。

    此外，VVR 4.0.8以下版本在全量讀取階段無法執行Checkpoint，全量階段的作業失敗會導致作業重新讀取全量數據，穩定性不佳。因此建議您將作業升級到VVR 4.0.8及以上版本。

• 結果表

  • RDS MySQL數據庫支持自增主鍵，因此在結果表的DDL中不聲明該自增字段。例如ID是自增字段，Flink DDL不聲明該自增字段，則數據庫在一行數據寫入過程中會自動填補相關自增字段。

  • 結果表的DDL聲明的字段必須至少存在一個非主鍵的字段，否則產生報錯。

  • 結果表的DDL中NOT ENFORCED表示Flink自身對主鍵不做強制校驗，需要您自行保證主鍵的正確性和完整性。

    Flink并不充分支持強制校驗，Flink將假設列的可為空性與主鍵中的列是對齊的，從而認為主鍵是正確的，詳情請參見Validity Check。

• 維表

  如果做維表時希望使用索引查詢，請按照MySQL最左前綴原則排列JOIN指定的數據列。但這并無法保證使用索引，由于SQL優化，某些條件可能會被優化導致連接器得到的過濾條件無法命中索引。要確定連接器是否真正使用了索引進行查詢，可以在數據庫側查看具體執行的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源表在啟動時掃描全表，將表按照主鍵分成多個分片（chunk），記錄下此時的Binlog位點。并使用增量快照算法通過select語句，逐個讀取每個分片的數據。作業會周期性執行Checkpoint，記錄下已經完成的分片。當發生Failover時，只需要繼續讀取未完成的分片。當分片全部讀取完后，會從之前獲取的Binlog位點讀取增量的變更記錄。Flink作業會繼續周期性執行Checkpoint，記錄下Binlog位點，當作業發生Failover，便會從之前記錄的Binlog位點繼續處理，從而實現Exactly Once語義。

  更詳細的增量快照算法，請參見MySQL CDC Connector。

• 元數據

  元數據在分庫分表合并同步場景非常實用，因為分庫分表合并后，一般業務還是希望區分每條數據的庫名和表名來源，而元數據列可以訪問源表的庫名和表名信息。因此通過元數據列可以非常方便地將多張分表合并到一張目的表。

  自vvr-4.0.11-flink-1.13版本開始，MySQL CDC Source支持元數據列語法，您可以通過元數據列訪問以下元數據。

  元數據key	元數據類型	描述
  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消息 說明 僅實時計算引擎VVR 8.0.7及以上版本支持。

  將MySQL實例中多個分庫下的多張orders表，合并同步到下游Hologres的holo_orders表中，代碼示例如下所示。

  CREATE 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).* 是前綴匹配示例，這個表達式可以匹配以test開頭的庫名，例如test1或test2。

  • .*[p$] 是后綴匹配示例， 這個表達式可以匹配以p結尾的庫名，例如cdcp或edcp。

  • 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實時計算控制臺的Autopilot自動調優功能，在多并發讀取完成后增量階段，能夠自動縮容，節約計算資源。

  在Flink全托管控制臺，您可以在資源配置頁面的基礎模式或專家模式中設置作業的并發數。設置并發的區別如下：

  • 基礎模式設置的并發數為整個作業的全局并發數。

  • 專家模式支持按需為某個VERTEX設置并發數。

  資源配置詳情請參見配置作業部署信息。

  重要

  無論是基礎模式還是專家模式，在設置并發時，表中聲明的server-id范圍必須大于等于作業的并發數。例如server-id的范圍為5404-5412，則共有8個唯一的server-id，因此作業最多可以設置8個并發，且不同的作業對于同一個MySQL實例的server-id范圍不能有重疊，即每個作業需顯式配置不同的server-id。

• Autopilot自動縮容

  全量階段積累了大量歷史數據，為了提高讀取效率，通常采用并發的方式讀取歷史數據。而在Binlog增量階段，因為Binlog數據量少且為了保證全局有序，通常只需要單并發讀取。全量階段和增量階段對資源的不同需求，可以通過自動調優功能自動幫您實現性能和資源的平衡。

  自動調優會監控MySQL CDC Source的每個task的流量。當進入Binlog階段，如果只有一個task在負責Binlog讀取，其他task均空閑時，自動調優便會自動縮小Source的CU數和并發。開啟自動調優只需要在作業運維頁面，將自動調優的模式設置為Active模式。

  說明

  默認調低并發度的最小觸發時間間隔為24小時。更多自動調優的參數和細節，請參見配置自動調優。

• 啟動模式

  使用配置項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', -- 在特定位點啟動模式下指定Binlog文件名。
      'scan.startup.specific-offset.pos' = '4', -- 在特定位點啟動模式下指定Binlog位置。
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- 在特定位點啟動模式下指定GTID集合。
      'scan.startup.timestamp-millis' = '1667232000000' -- 在時間戳啟動模式下指定啟動時間戳。
      ...
  )

  重要

  • MySQL source會在Checkpoint時將當前位點以INFO級別打印到日志中，日志前綴為Binlog offset on checkpoint {checkpoint-id}，該日志可以幫助您將作業從某個Checkpoint位點開始啟動作業。

  • 如果讀取的表曾經發生過表結構變化，從最早位點（earliest-offset）、特定位點（specific-offset）或時間戳（timestamp）啟動可能會發生錯誤。因為Debezium讀取器會在內部保存當前的最新表結構，結構不匹配的早期數據無法被正確解析。

• 關于無主鍵CDC源表

  • 在Flink計算引擎VVR 6.0.7及以上版本支持使用MySQL CDC無主鍵源表，使用無主鍵表要求必須設置scan.incremental.snapshot.chunk.key-column，且只能選擇非空類型的字段。

  • 無主鍵CDC源表的處理語義由scan.incremental.snapshot.chunk.key-column指定的列的行為決定：

    • 如果指定的列不存在更新操作，此時可以保證Exactly once語義。

    • 如果指定的列發生更新操作，此時只能保證At least once語義。但可以結合下游，通過指定下游主鍵，結合冪等性操作來保證數據的正確性。

• 讀取阿里云RDS MySQL備份日志

  MySQL CDC源表支持讀取阿里云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 Source復用

  當同一個作業中有多個MySQL CDC源表時，每個源表都會啟動對應的Binlog Client，如果源表數量較多并且讀取的MySQL表都在同一個實例中時，會對數據庫造成較大壓力，詳情請參見MySQL CDC常見問題。

  實時計算引擎VVR 8.0.7及以上版本支持MySQL CDC Source復用，當不同的CDC源表配置項除了數據庫、表名和server-id外的其他配置項均相同時，可以進行合并。開啟Source復用后，實時計算引擎會盡可能將同一個作業中能夠合并的MySQL CDC源表進行合并。

  您可以在SQL作業中使用SET命令開啟source復用功能：

  SET 'table.optimizer.source-merge.enabled' = 'true';

  對已有作業啟用 Source 復用后，需要無狀態啟動。原因是 Source 復用會導致作業拓撲改變，從原有作業狀態可能無法啟動或者丟失數據。

  重要

  • VVR 8.0.8及8.0.9版本，在開啟CDC Source復用時，還需要額外設置SET 'sql-gateway.exec-plan.enabled' = 'false'。

  • 在開啟CDC Source復用后，不建議將作業配置項pipeline.operator-chaining設為false，因為將算子鏈斷開后，Source發送給下游算子的數據會增加序列化和反序列的開銷，當合并的Source越多時，開銷會越大。

  • 在實時計算引擎VVR 8.0.7版本，將pipeline.operator-chaining設為false時會出現序列化的問題。

加速Binlog讀取

MySQL連接器作為源表或數據攝入數據源使用時，在增量階段會解析Binlog文件生成各種變更消息，Binlog文件使用二進制記錄著所有表的變更，可以通過以下方式加速Binlog文件解析。

• 開啟并行解析和解析過濾配置

  • 使用配置項scan.only.deserialize.captured.tables.changelog.enabled：僅對指定表的變更事件進行解析。

  • 使用配置項scan.parallel-deserialize-changelog.enabled：采用多線程對Binlog文件進行解析、并按順序投放到消費隊列。

• 優化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',  -- 僅對指定表的變更事件進行解析。
    'scan.parallel-deserialize-changelog.enabled' = 'true'  -- 使用多線程對Binlog進行解析。
    ...
)

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
  scan.parallel-deserialize-changelog.enabled: true

MySQL CDC 企業版本binlog消費能力為85MB/s，約為開源社區的2倍，當Binlog文件產生速度大于 85MB/s 時（即每6s一個512MB大小的文件），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") // 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問題。

Flink CDC技術原理及企業版特性

• Flink CDC企業版特性

• Flink CDC技術