AnalyticDB：Flink訂閱Binlog

前提條件

• AnalyticDB for MySQL產品系列為企業版、基礎版、湖倉版和數倉版彈性模式。

• AnalyticDB for MySQL叢集的核心版本：

  • xuanwu_v1表引擎：需為3.2.1.0及以上版本。

  • xuanwu_v2表引擎：需為3.2.6.0及以上版本。

  • 如需訂閱物化視圖增量重新整理的Binlog，核心版本需為3.2.6.9、3.2.7.1及以上。

  說明

  請在雲原生資料倉儲AnalyticDB MySQL控制台集群信息頁面，配寘資訊地區，查看和升級核心版本。

• FlinkRealtime Compute引擎需為VVR 8.0.4及以上版本。

• AnalyticDB for MySQL叢集和Flink全託管工作空間需要位於同一VPC下。

• 已將Flink工作空間所屬的網段加入AnalyticDB for MySQL的白名單。

使用限制

• Flink僅支援處理AnalyticDB for MySQL Binlog中的所有基礎資料類型和複雜資料類型JSON。

• 使用Flink訂閱AnalyticDB for MySQL Binlog時，以下操作產生的變更不會被捕獲，相應資料不會同步到下遊：

  • DDL操作：如ALTER TABLE等表結構變更。如果源表發生結構變更，需要手動更新Flink作業的表定義並重新部署。

  • INSERT OVERWRITE：使用INSERT OVERWRITE批量寫入的資料不會產生Binlog記錄。如果下遊需要感知INSERT OVERWRITE後的完整資料，需要重新訂閱Binlog。

  • 分區表自動淘汰：分區表中到期分區被自動淘汰時，淘汰的資料不會產生Binlog記錄。

  • 物化視圖全量重新整理：對物化視圖執行全量重新整理時不會產生Binlog記錄。

    說明

    物化視圖的Binlog僅包含增量重新整理產生的變更記錄。如需感知全量重新整理後的完整資料，需要重新訂閱Binlog。

步驟三：訂閱Binlog

1. 登入Realtime Compute控制台，建立SQL作業。

2. 建立源表，串連到AnalyticDB for MySQL並讀取指定表（source_table）的Binlog資料。

   說明

   • Flink DDL中定義的主鍵必須和AnalyticDB for MySQL叢集物理表中的主鍵保持一致，主鍵一致包括主鍵和主鍵名稱一致。如果不一致，會影響資料正確性。

   • Flink的資料類型需要和AnalyticDB for MySQL相容。映射關係，請參見類型映射。

   CREATE TEMPORARY TABLE adb_source (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
     'connector' = 'adb-mysql-cdc',
     'hostname' = 'amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com',
     'username' = 'testUser',
     'password' = 'Test12****',
     'database-name' = 'binlog',
     'table-name' = 'source_table'
   );

   WITH參數說明：

   參數	是否必填	預設值	資料類型	說明
   connector	是	無	STRING	使用的連接器。 這裡填寫自訂連接器，固定填寫adb-mysql-cdc。
   hostname	是	無	STRING	AnalyticDB for MySQL的VPC地址。
   username	是	無	STRING	AnalyticDB for MySQL資料庫帳號。
   Password	是	無	STRING	AnalyticDB for MySQL資料庫密碼。
   database-name	是	無	STRING	AnalyticDB for MySQL資料庫名稱。 由於AnalyticDB for MySQL實現的是表級Binlog，此處僅支援設定一個資料庫。
   table-name	是	無	STRING	AnalyticDB for MySQL資料庫的表名。 由於AnalyticDB for MySQL實現的是表級Binlog，此處僅支援設定一個表。
   Port	否	3306	INTEGER	連接埠號碼。
   scan.incremental.snapshot.enabled	否	true	BOOLEAN	增量快照。 預設開啟。增量快照是一種讀取錶快照的新機制，與舊的快照機制相比，增量快照有許多優點，包括： 在讀取快照期間，Source支援並發讀取。 在讀取快照期間，Source支援進行Chunk粒度的Checkpoint。 在讀取快照之前，Source不需要擷取資料庫鎖許可權。
   scan.incremental.snapshot.chunk.size	否	8096	INTEGER	錶快照的Chunk大小（包含的行數）。 當開啟增量快照讀取時，表會被切分成多個Chunk讀取。
   scan.snapshot.fetch.size	否	1024	INTEGER	讀取錶快照時，每次讀取資料的最大行數。
   scan.startup.mode	否	initial	STRING	消費資料的啟動模式。 取值如下： initial（預設）：在第一次啟動時，會先掃描歷史全量資料，然後讀取最新的Binlog資料。 earliest-offset：不掃描歷史全量資料，直接從可讀取的最早Binlog開始讀取。 specific-offset：不掃描歷史全量資料，從您指定的Binlog位點啟動，位點可通過同時配置scan.startup.specific-offset.file和scan.startup.specific-offset.pos參數來指定從特定Binlog檔案名稱和位移量啟動。 latest-offset：在第一次啟動時，不會掃描歷史全量資料，直接從Binlog的末尾（最新的Binlog處）開始讀取，即唯讀取該連接器啟動以後的最新變更。 timestamp：不掃描歷史全量資料，從指定的時間戳記開始讀取Binlog。時間戳記通過scan.startup.timestamp-millis指定，單位為毫秒（ms）。 重要 使用earliest-offset、specific-offset或timestamp啟動模式時，請確保在指定的Binlog消費位置到作業啟動期間，對應表的結構保持不變，避免因表結構變更導致作業執行失敗。
   scan.startup.specific-offset.file	否	無	STRING	在specific-offset啟動模式下，啟動位點的Binlog檔案名稱。 最新Binlog檔案名稱可使用SHOW MASTER STATUS for table_name;擷取。
   scan.startup.specific-offset.pos	否	無	LONG	在specific-offset啟動模式下，啟動位點的Binlog檔案位置。 最新Binlog位置可使用SHOW MASTER STATUS for table_name;擷取。
   scan.startup.specific-offset.skip-events	否	無	LONG	在指定的啟動位點後需要跳過的事件數目量。
   scan.startup.specific-offset.skip-rows	否	無	LONG	在指定的啟動位點後需要跳過的資料行數。
   scan.startup.timestamp-millis	否	無	LONG	使用指定時間模式啟動時，啟動位點的毫秒時間戳記。 使用該配置時，scan.startup.mode必須配置為timestamp。時間戳記單位為毫秒（ms）。
   server-time-zone	否	無	STRING	資料庫伺服器中的會話時區。 例如："Asia/Shanghai"。它控制AnalyticDB for MySQL中的TIMESTAMP類型如何轉成STRING類型。如果沒有設定，則使用ZONELD.SYSTEMDEFAULT()來確定伺服器時區。
   debezium.min.row.count.to.stream.result	否	1000	INTEGER	當表的行數大於該值時，連接器會對結果進行串流。 若將此參數設定為0，會跳過所有表大小檢查，始終在快照期間對所有結果進行串流。
   connect.timeout	否	30s	DURATION	串連資料庫伺服器逾時，重試串連之前等待逾時的最長時間。 預設單位為秒（s）。
   connect.max-retries	否	3	INTEGER	串連資料庫服務時，串連失敗後重試的最大次數。

3. 在目標端建立表，用於儲存處理後的資料。本文以AnalyticDB for MySQL作為目標端。Flink支援的連接器請參見支援的連接器。

   CREATE TABLE target_table (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`)
   )

4. 建立結果表，串連步驟3建立的表，用於將處理後的資料寫入到AnalyticDB for MySQL指定的表。

   CREATE TEMPORARY TABLE adb_sink (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
     'connector' = 'adb3.0',
     'url' = 'jdbc:mysql://amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com:3306/flinktest',
     'userName' = 'testUser',
     'password' = 'Test12****',
     'tableName' = 'target_table'
   );

   結果表對應的WITH參數和映射類型的詳情，請參見：雲原生資料倉儲AnalyticDB MySQL版（ADB）3.0。

5. 將捕獲到的來源資料變化同步到結果表，並由結果表將資料同步到目標端。

   INSERT INTO adb_sink
   SELECT * FROM adb_source;

6. 單擊保存。

7. 單擊深度检查。

   深度檢查能夠檢查作業的SQL語義、網路連通性以及作業使用的表的中繼資料資訊。同時，您可以單擊結果地區的SQL最佳化，展開查看SQL風險問題提示以及對應的SQL最佳化建議。

8. （可選）單擊調試。

   您可以使用作業調試功能類比作業運行、檢查輸出結果，驗證SELECT或INSERT商務邏輯的正確性，提升開發效率，降低資料品質風險。

9. 單擊部署。

   完成作業開發和深度檢查後，即可部署作業，將資料發布至生產環境。部署後，您可以在作業營運頁面啟動作業至運行階段。

10. （可選）查看Binlog資訊。

    說明

    使用以下語句查看Binlog日誌資訊時，若僅開啟Binlog功能，日誌資訊顯示為0。只有成功訂閱Binlog後，才會顯示日誌資訊。

    • 若您需要擷取Binlog最新寫入的檔案名稱和位置資訊，請執行以下SQL語句：

      SHOW MASTER STATUS FOR source_table;

    • 若您需要瞭解所有未清理的歷史Binlog檔案及其大小，請執行以下SQL語句：

      SHOW BINARY LOGS FOR source_table;

類型映射

AnalyticDB for MySQL與Flink的資料類型映射關係如下：