Realtime Compute for Apache Flink：MySQL

背景資訊

MySQL連接器支援所有相容MySQL協議的資料庫，包括RDS MySQL、PolarDB for 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。

  • 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訊息 說明 僅Realtime Compute引擎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時，輸出結果包含上遊表每個主鍵的完整變更。

• 支援Regex

  MySQL CDC源表支援在表名或者庫名中使用Regex匹配多個表或者多個庫。通過Regex指定多張表的程式碼範例如下。

  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})', -- Regex匹配多個庫。
    'table-name' = '(t[5-8]|tt)' -- Regex匹配多張表。
  );

  上述例子中的Regex解釋：

  • ^(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。

  重要

  table-name和database-name也支援使用逗號（,）分隔形式指定多張表或多個庫，例如 'table-name' = 'mytable1,mytable2'。但是其與Regex中的逗號（,）衝突，所以如果使用了含有逗號的Regex，則Regex需要改寫成豎線（|）的形式，例如mytable_\d{1, 2)需要改寫成等價的Regex(mytable_\d{1}|mytable_\d{2})，來避免使用逗號。

• 並發控制

  MySQL連接器支援多並發讀取全量資料，能夠提高資料載入效率。同時配合FlinkRealtime Compute控制台的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常見問題。

  Realtime Compute引擎VVR 8.0.7及以上版本支援MySQL CDC Source複用，當不同的CDC源表配置項除了資料庫、表名和server-id外的其他配置項均相同時，可以進行合并。開啟Source複用後，Realtime Compute引擎會儘可能將同一個作業中能夠合并的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越多時，開銷會越大。

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

加速Binlog讀取

MySQL連接器作為源表或資料攝入資料來源使用時，在增量階段會解析Binlog檔案產生各種變更訊息，Binlog檔案使用二進位記錄著所有表的變更，可以通過以下方式加速Binlog檔案解析。

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

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

  • 使用配置項scan.only.deserialize.captured.tables.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問題。