操作原理與Demo演示

• Flink CDC如何增強(qiáng)和簡(jiǎn)化實(shí)時(shí)數(shù)據(jù)入倉

• MySQL實(shí)時(shí)同步至Hologres

背景信息

本文使用Flink提供的CREATE TABLE AS（CTAS）語句和CREATE DATABASE AS（CDAS）語句來完成整庫同步、分庫分表合并同步，一鍵完成數(shù)據(jù)的全量和增量同步，以及實(shí)時(shí)的表結(jié)構(gòu)變更同步。

前提條件

• 如果您使用RAM用戶或RAM角色等身份訪問，需要確認(rèn)已具有Flink控制臺(tái)相關(guān)權(quán)限，詳情請(qǐng)參見權(quán)限管理。

• 已創(chuàng)建Flink工作空間，詳情請(qǐng)參見開通實(shí)時(shí)計(jì)算Flink版。

• 上下游存儲(chǔ)

  • 已創(chuàng)建RDS MySQL實(shí)例，詳情請(qǐng)參見快速創(chuàng)建RDS MySQL實(shí)例。

  • 已創(chuàng)建Hologres實(shí)例，詳情請(qǐng)參見購買Hologres。

  說明

  RDS MySQL和Hologres需要與Flink工作空間在相同地域相同VPC下，否則需要打通網(wǎng)絡(luò)，詳情請(qǐng)參見控制臺(tái)操作或控制臺(tái)操作。

• 已準(zhǔn)備好測(cè)試數(shù)據(jù)，并配置好白名單。詳情請(qǐng)參見準(zhǔn)備測(cè)試數(shù)據(jù)和配置IP白名單。

準(zhǔn)備測(cè)試數(shù)據(jù)

1. 單擊tpc_ds.sql、user_db1.sql、user_db2.sql和user_db3.sql下載測(cè)試數(shù)據(jù)到本地。

2. 在DMS數(shù)據(jù)管理控制臺(tái)上，準(zhǔn)備RDS MySQL的測(cè)試數(shù)據(jù)。

   1. 通過DMS登錄RDS MySQL。

      詳情請(qǐng)參見通過DMS登錄RDS MySQL。

   2. 在已登錄的SQLConsole窗口，輸入如下命令后單擊執(zhí)行。

      創(chuàng)建tpc_ds、user_db1、user_db2和user_db3四個(gè)數(shù)據(jù)庫。

      CREATE DATABASE tpc_ds;
      CREATE DATABASE user_db1;
      CREATE DATABASE user_db2;
      CREATE DATABASE user_db3;

   3. 在左側(cè)菜單欄，單擊左側(cè)常用功能頁簽下的數(shù)據(jù)導(dǎo)入。

   4. 在批量數(shù)據(jù)導(dǎo)入頁簽下選擇需要導(dǎo)入的數(shù)據(jù)庫，上傳對(duì)應(yīng)的SQL文件，單擊提交申請(qǐng)后，單擊執(zhí)行變更。在彈出的對(duì)話框中單擊確定執(zhí)行。

      同樣的操作依次為tpc_ds、user_db1、user_db2和user_db3數(shù)據(jù)庫導(dǎo)入對(duì)應(yīng)的數(shù)據(jù)文件。

3. 在Hologres控制臺(tái)創(chuàng)建my_user數(shù)據(jù)庫，用于存放合并后的user表數(shù)據(jù)。

   操作步驟詳情請(qǐng)參見創(chuàng)建數(shù)據(jù)庫。

配置IP白名單

為了讓Flink能訪問MySQL和Hologres實(shí)例，您需要將Flink工作空間的網(wǎng)段添加到MySQL和Hologres的白名單中。

1. 獲取Flink工作空間的VPC網(wǎng)段。

   1. 登錄實(shí)時(shí)計(jì)算控制臺(tái)。

   2. 在目標(biāo)工作空間右側(cè)操作列，選擇更多 > 工作空間詳情。

   3. 在工作空間詳情對(duì)話框，查看Flink虛擬交換機(jī)的網(wǎng)段信息。

      

2. 在RDS MySQL的IP白名單中，添加Flink網(wǎng)段信息。

   操作步驟詳情請(qǐng)參見設(shè)置IP白名單。

3. 在Hologres的IP白名單中，添加Flink網(wǎng)段信息。

   在HoloWeb配置數(shù)據(jù)連接時(shí)，需要將連接的登錄方式設(shè)置為當(dāng)前用戶免密登錄，才可以為當(dāng)前連接配置IP白名單，操作步驟詳情請(qǐng)參見IP白名單。

步驟一：創(chuàng)建Catalog

整庫同步、分庫分表合并同步、單表同步都需要依賴目標(biāo)Catalog來創(chuàng)建目標(biāo)表，也依賴源Catalog來獲取源表列表和信息。因此，您需要通過控制臺(tái)創(chuàng)建源Catalog和目標(biāo)Catalog。本文將以源Catalog為MySQL Catalog和目標(biāo)Catalog為Hologres Catalog為例，為您進(jìn)行介紹。

1. 創(chuàng)建名稱為mysql的MySQL Catalog。

   操作步驟詳情請(qǐng)參見配置MySQL Catalog。

2. 創(chuàng)建名稱為holo的Hologres Catalog。

   操作步驟詳情請(qǐng)參見創(chuàng)建Hologres Catalog。

3. 在元數(shù)據(jù)管理頁面Catalog列表中，確認(rèn)已創(chuàng)建名為mysql和holo的Catalog。

步驟二：開發(fā)數(shù)據(jù)同步作業(yè)

1. 登錄Flink開發(fā)控制臺(tái)，新建作業(yè)。

   1. 在數(shù)據(jù)開發(fā) > ETL頁面，單擊新建。

   2. 單擊空白的流作業(yè)草稿。

      Flink為您提供了豐富的代碼模板和數(shù)據(jù)同步，每種代碼模板都為您提供了具體的使用場(chǎng)景、代碼示例和使用指導(dǎo)。您可以直接單擊對(duì)應(yīng)的模板快速地了解Flink產(chǎn)品功能和相關(guān)語法，實(shí)現(xiàn)您的業(yè)務(wù)邏輯，詳情請(qǐng)參見代碼模板和數(shù)據(jù)同步模板。

   3. 單擊下一步。

   4. 在新建作業(yè)草稿對(duì)話框，填寫作業(yè)配置信息。

      作業(yè)參數(shù)	說明	示例
      文件名稱	作業(yè)的名稱。 說明 作業(yè)名稱在當(dāng)前項(xiàng)目中必須保持唯一。	flink-test
      存儲(chǔ)位置	指定該作業(yè)的代碼文件所屬的文件夾。 您還可以在現(xiàn)有文件夾右側(cè)，單擊圖標(biāo)，新建子文件夾。	作業(yè)草稿
      引擎版本	當(dāng)前作業(yè)使用的Flink的引擎版本。引擎版本號(hào)含義、版本對(duì)應(yīng)關(guān)系和生命周期重要時(shí)間點(diǎn)詳情請(qǐng)參見引擎版本介紹。	vvr-6.0.4-flink-1.15

   5. 單擊創(chuàng)建。

2. 將以下作業(yè)代碼拷貝到作業(yè)文本編輯區(qū)。

   將tpc_ds庫中所有表同步至Hologres，并將user的分庫分表合并同步到Hologres的單表中。代碼示例如下所示。

   USE CATALOG holo;
   
   BEGIN STATEMENT SET;
   
   -- 同步TPCDS整庫到Hologres的tpc_ds庫中。
   CREATE DATABASE IF NOT EXISTS tpc_ds
   AS DATABASE mysql.tpc_ds INCLUDING ALL TABLES
   /*+ OPTIONS('server-id'='8001-8004') */ ;
   
   -- 同步user分庫分表到Hologres的my_user.users表中。
   CREATE TABLE IF NOT EXISTS my_user.users
   AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
   /*+ OPTIONS('server-id'='8001-8004') */;
   
   END;

   將tpc_ds庫中所有表同步至Hologres使用CDAS (CREATE DATABASE AS) 語法來實(shí)現(xiàn)，將user的分庫分表合并同步到Hologres的單表使用CTAS (CREATE TABLE AS) 語法來實(shí)現(xiàn)，最后再使用STATEMENT SET語法將這兩條SQL語句合并在一個(gè)作業(yè)中提交。Flink會(huì)自動(dòng)為Source進(jìn)行優(yōu)化，復(fù)用一個(gè)Source節(jié)點(diǎn)讀取多張MySQL表的數(shù)據(jù)，這能顯著降低MySQL的連接數(shù)和讀取壓力，提升穩(wěn)定性。

   說明

   如果只想同步庫中的某些表，您也可以在CDAS語法中使用 INCLUDING TABLE或EXCLUDING TABLE 語法來指定具體需要同步的表。例如INCLUDING TABLE 'web.*'表示只同步中所有web開頭的表。

步驟三：?jiǎn)?dòng)作業(yè)

1. 在數(shù)據(jù)開發(fā) > ETL頁面，單擊部署后，在彈出的對(duì)話框中，單擊確認(rèn)。

   說明

   Session集群適用于非生產(chǎn)環(huán)境的開發(fā)測(cè)試環(huán)境，您可以使用Session集群模式部署或調(diào)試作業(yè)，提高作業(yè)JM（Job Manager）資源利用率和提高作業(yè)啟動(dòng)速度。但不推薦您將作業(yè)提交至Session集群中，因?yàn)闀?huì)存在業(yè)務(wù)穩(wěn)定性問題，詳情請(qǐng)參見配置開發(fā)測(cè)試環(huán)境（Session集群）。

2. 在運(yùn)維中心 > 作業(yè)運(yùn)維頁面，單擊目標(biāo)作業(yè)操作中的啟動(dòng)。填寫配置信息，詳情請(qǐng)參見作業(yè)啟動(dòng)。

3. 單擊啟動(dòng)。

   作業(yè)啟動(dòng)后，您可以在作業(yè)運(yùn)維頁面觀察作業(yè)的運(yùn)行信息和狀態(tài)。

步驟四：觀察全量同步結(jié)果

1. 登錄Hologres管理控制臺(tái)。

2. 在元數(shù)據(jù)管理頁簽，查看Hologres實(shí)例下的tpc_ds數(shù)據(jù)庫中24張表和表數(shù)據(jù)。

   

3. 在元數(shù)據(jù)管理頁簽，查看my_user庫下users表結(jié)構(gòu)。

   同步后的表結(jié)構(gòu)和數(shù)據(jù)如下圖所示。

   • 表結(jié)構(gòu)

     users表的表結(jié)構(gòu)比MySQL源表中多了_db_name和_table_name兩列，代表數(shù)據(jù)來源的庫名和表名，且作為聯(lián)合主鍵的一部分來保證分庫分表合并后的數(shù)據(jù)唯一性。

   • 表數(shù)據(jù)

     在users表信息頁面右上角，單擊查詢表后，輸入如下命令，單擊運(yùn)行。

     select * from users order by _db_name,_table_name,id;

     表數(shù)據(jù)結(jié)果如下圖所示。

步驟五：觀察增量同步結(jié)果

同步作業(yè)會(huì)在全量數(shù)據(jù)同步完以后自動(dòng)切換到增量數(shù)據(jù)同步階段，無需干預(yù)。您可以通過監(jiān)控告警頁簽的currentEmitEventTimeLag值來確定數(shù)據(jù)同步的階段。

1. 登錄實(shí)時(shí)計(jì)算控制臺(tái)。

2. 單擊對(duì)應(yīng)工作空間操作列下的控制臺(tái)。

3. 在運(yùn)維中心 > 作業(yè)運(yùn)維頁面，單擊目標(biāo)作業(yè)名稱。

4. 單擊監(jiān)控告警頁簽。

5. 觀察currentEmitEventTimeLag曲線圖，確定數(shù)據(jù)同步階段。

   

   • 值為0時(shí)，代表還在全量同步階段。

   • 值大于0時(shí)，代表已經(jīng)進(jìn)入增量同步階段。

6. 驗(yàn)證實(shí)時(shí)同步數(shù)據(jù)變更和結(jié)構(gòu)變更的能力。

   MySQL CDC數(shù)據(jù)源支持在增量同步階段，實(shí)時(shí)同步表的數(shù)據(jù)變更以及表的結(jié)構(gòu)變更。您可以在作業(yè)進(jìn)入到增量同步階段后，通過修改MySQL的user分表的表結(jié)構(gòu)和數(shù)據(jù)，來驗(yàn)證實(shí)時(shí)同步數(shù)據(jù)變更和結(jié)構(gòu)變更的能力。

   1. 通過DMS登錄RDS MySQL。

      詳情請(qǐng)參見通過DMS登錄RDS MySQL。

   2. 在user_db2數(shù)據(jù)庫下，執(zhí)行如下命令修改user02表的表結(jié)構(gòu)，并插入和更新數(shù)據(jù)。

      USE DATABASE `user_db2`;
      ALTER TABLE `user02` ADD COLUMN `age` INT;   -- 添加age列。
      INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30); -- 插入帶有age的數(shù)據(jù)。
      UPDATE `user05` SET name='JARK' WHERE id=15;  -- 更新另一張表，名字改成大寫。

   3. 在Hologres控制臺(tái)，查看users表結(jié)構(gòu)和數(shù)據(jù)的變化。

      在users表信息頁面右上角，單擊查詢表后，輸入如下命令，單擊運(yùn)行。

      select * from users order by _db_name,_table_name,id;

      表數(shù)據(jù)結(jié)果如下圖所示。雖然多張分表的Schema并不一致，但是在user02上的表結(jié)構(gòu)變更，以及數(shù)據(jù)變更都能實(shí)時(shí)地同步到下游表中。在Hologres的users表中，看到了新增的age字段，插入的Tony數(shù)據(jù)以及更新成大寫的JARK數(shù)據(jù)。

（可選）步驟六：作業(yè)資源配置

根據(jù)數(shù)據(jù)量的不同，我們往往需要調(diào)節(jié)不同節(jié)點(diǎn)的并發(fā)和資源，以達(dá)到更優(yōu)的作業(yè)性能。您可以使用資源配置的基礎(chǔ)模式簡(jiǎn)單配置作業(yè)并發(fā)度和CU數(shù)，也可以使用資源配置的專家模式細(xì)粒度地調(diào)整節(jié)點(diǎn)的并發(fā)和資源。

1. 在運(yùn)維中心 > 作業(yè)運(yùn)維頁面，單擊目標(biāo)作業(yè)名稱。

2. 在部署詳情頁簽下，單擊資源配置區(qū)域右上角的編輯。

3. 在資源模式配置項(xiàng)中選擇專家模式后，單擊立刻獲取。

4. 將鼠標(biāo)懸停至更多操作后，單擊展開全部。

   通過完整的拓?fù)鋱D能了解到整個(gè)數(shù)據(jù)的同步計(jì)劃，即具體同步哪些表。

5. 手動(dòng)設(shè)置每個(gè)節(jié)點(diǎn)的并發(fā)。

   設(shè)置作業(yè)為4并發(fā)；由于tpc_ds中的store_sales表數(shù)據(jù)量最大，可以單獨(dú)設(shè)置holo.tpc_ds.store_sales Sink節(jié)點(diǎn)并發(fā)為8，提升Hologres的寫入性能。資源配置步驟詳情請(qǐng)參見配置作業(yè)部署信息。

6. 在資源配置右側(cè)，單擊保存。

7. 重啟作業(yè)。

   作業(yè)資源配置后，需重啟作業(yè)才能生效。

8. 單擊目標(biāo)作業(yè)名稱，在作業(yè)總覽頁簽下查看調(diào)整效果。

常見問題

• 更新了MySQL中的表結(jié)構(gòu)，但是下游的表結(jié)構(gòu)沒有變化是怎么回事？

• Source出現(xiàn)finish split response timeout異常，是什么原因？

• 在MySQL CDC全量階段發(fā)生表結(jié)構(gòu)變更有什么影響？

• 如果CTAS/CDAS同步期間發(fā)生了不支持的表結(jié)構(gòu)變更，導(dǎo)致作業(yè)報(bào)錯(cuò)同步失敗，該怎么解決？

相關(guān)文檔

• CREATE TABLE AS (CTAS) 語法功能介紹，請(qǐng)參見CREATE TABLE AS（CTAS）語句。

• CREATE DATABASE AS (CDAS) 語法功能介紹，請(qǐng)參見CREATE DATABASE AS（CDAS）語句。

• 日志實(shí)時(shí)入倉快速入門。

• 基于Paimon的數(shù)據(jù)庫實(shí)時(shí)入湖快速入門。

• 基于Flink+Hologres搭建實(shí)時(shí)數(shù)倉。

• 基于Flink+Paimon搭建流式湖倉。