Realtime Compute for Apache Flink：MySQL

背景資訊

MySQL連接器支援所有相容MySQL協議的資料庫，包括RDS MySQL、PolarDB for MySQL、OceanBase（MySQL模式）或者自建MySQL。

特色功能

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

前提條件

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

使用限制

注意事項

• 每個MySQL CDC資料來源需顯式配置不同的Server ID

  Server ID作用

  每個MySQL CDC資料來源需顯式配置不同的Server ID。如果多個MySQL CDC資料來源共用相同的Server ID，且資料來源之間無法複用時，會導致Binlog位點錯亂，資料多讀或少讀。

  不同情境下Server ID的配置

  Server ID可以在DDL中指定，但建議通過動態Hints來配置Server ID，而不是在DDL參數中配置Server ID。

  • 並行度 = 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。詳情請參見程式碼範例四：多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會自動填寫。

  • 必須至少聲明一個非主鍵欄位，否則報錯。

  • DDL中NOT ENFORCED表示Flink自身對主鍵不做強制校正，需要您自行保證主鍵的正確性和完整性。詳情請參見Validity Check。

• 維表

  如果希望使用索引加速查詢，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 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。

• 中繼資料

  中繼資料在分庫分表合并同步情境非常實用，因為分庫分表合并後，一般業務還是希望區分每條資料的庫名和表名來源，而中繼資料列可以訪問源表的庫名和表名資訊。因此通過中繼資料列可以非常方便地將多張分表合并到一張目的表。

  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及以上版本支援。
  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時，輸出結果包含上遊表每個主鍵的完整變更。

• 支援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。

  重要

  在SQL作業的配置中，table-name和database-name不支援使用逗號（,）分隔形式指定多張表或多個庫。

  • 如果需要匹配多個表或使用多個Regex，可以用豎線（|）串連並用小括弧包圍，例如需要讀取表user和product，table-name可以配置為(user|product)。

  • 如果Regex包含逗號，需要用豎線（|）運算子進行改寫，例如Regexmytable_\d{1, 2}需要改寫成等價的(mytable_\d{1}|mytable_\d{2})，來避免使用逗號。

• 並發控制

  MySQL連接器支援多並發讀取全量資料，能夠提高資料載入效率。同時配合FlinkRealtime Compute控制台的Autopilot自動調優功能，在多並發讀取完成後增量階段，能夠自動縮容，節約計算資源。

  在Realtime Compute開發控制台，您可以在資源配置頁面的基礎模式或專家模式中設定作業的並發數。設定並發的區別如下：

  • 基礎模式設定的並發數為整個作業的全域並發數。

  • 專家模式支援按需為某個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源表

  • 使用無主鍵表要求必須設定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 用戶端。所有源表都在同一執行個體時，這會增加資料庫壓力。詳情請參見MySQL CDC常見問題。

  解決方案

  Realtime Compute引擎 VVR 8.0.7 及以上版本支援 MySQL CDC Source 複用。複用會合并可合并的 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. 無狀態啟動作業。 修改 Source 複用配置項會改變作業拓撲。你必須無狀態啟動作業。否則，作業可能無法啟動或遺失資料。如果存在 Source 被合并，你會看到一個 MergetableSourceScan 節點。

  重要

  • 開啟複用後不建議禁用運算元鏈。如果將 pipeline.operator-chaining 設為 false，會增加資料序列化和還原序列化的開銷。合并的 Source 越多，開銷越大。

  • 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：該連接器每次迭代處理的事件條數最大值。預設值為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 企業版本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問題。