E-MapReduce：基於Realtime ComputeFlink使用CTAS語句同步MySQL資料至StarRocks

前提條件

• 已開通阿里雲Realtime ComputeFlink全託管並建立了Flink叢集，詳情請參見開通Flink全託管和Flink SQL作業快速入門。

• 已建立EMR Serverless StarRocks執行個體，詳情請參見建立執行個體。

• 已建立RDS MySQL，詳情請參見建立RDS MySQL執行個體。

使用限制

• 建立的Flink叢集、EMR Serverless StarRocks執行個體以及RDS MySQL執行個體需要在同一個VPC下。

• RDS MySQL須為5.7及以上版本。

步驟一：準備測試資料

1. 建立測試的資料庫和帳號，詳情請參見建立資料庫和帳號。

   建立完資料庫和帳號後，需要授權測試帳號的讀寫權限。

   說明

   本文建立的資料庫名稱為test_cdc，帳號為test。

2. 使用建立的測試帳號串連MySQL執行個體，詳情請參見通過DMS登入RDS MySQL。

3. 在MySQL中執行以下命令，建立資料表。

   use test_cdc;
   -- 建立表
   CREATE TABLE IF NOT EXISTS `runoob_tbl`(
      `runoob_id` INT UNSIGNED AUTO_INCREMENT,
      `runoob_title` VARCHAR(100) NOT NULL,
      `runoob_author` VARCHAR(40) NOT NULL,
      `submission_date` DATE,
      `add_col` int DEFAULT NULL,
      PRIMARY KEY ( `runoob_id` )
   )ENGINE=InnoDB DEFAULT CHARSET=utf8;
   
   -- 插入資料
   INSERT INTO test_cdc.`runoob_tbl` (`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`) values (18,'first','tom','2025-06-22 17:13:44',3)

4. 登入並串連EMR Serverless StarRocks執行個體，詳情請參見通過MySQL用戶端方式串連StarRocks執行個體。

5. 執行以下命令，建立資料庫test_cdc、建立超級管理使用者test（樣本密碼為1qaz!QAZ）或者建立普通使用者test並給普通使用者授予該資料庫及表的許可權，詳情請參見系統管理使用者。

   -- 建立資料庫
   CREATE DATABASE test_cdc;
   -- 建立使用者
   CREATE USER 'test' IDENTIFIED by '1qaz!QAZ';
   -- 給使用者授權資料庫許可權
   GRANT ALL on test_cdc to test;
   -- 給使用者授權表許可權
   GRANT ALL ON ALL TABLES IN DATABASE test_cdc to test;

步驟二：在Realtime ComputeFlink控制台通過SQL用戶端建立Catalog

• MySQL Catalog

  • 程式碼範例

    CREATE CATALOG mysql WITH (
      'type' = 'mysql',
      'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
      'port' = '3306',
      'username' = 'emr-test',
      'password' = '123456',
      'default-database' = 'test_cdc'
    );

  • 參數配置

    參數	說明
    type	類型，固定值為mysql。
    hostname	RDS的內網地址。您可以在RDS的資料庫連接頁面，單擊內網地址進行複製。例如，rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com。
    port	MySQL資料庫服務的連接埠號碼，預設值為3306。
    username	MySQL資料庫服務的使用者名稱。 填寫步驟一：準備測試資料中帳號的使用者名稱。本樣本為test。
    password	MySQL資料庫服務的密碼。 填寫步驟一：準備測試資料中帳號的密碼。
    default-database	預設的MySQL資料庫名稱。 填寫步驟一：準備測試資料中建立的資料庫名。本樣本為test_cdc。

• StarRocks Catalog

  • 程式碼範例

    CREATE CATALOG sr  WITH (
      'type' = 'starrocks',
      'endpoint' = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
      'username' = 'test',
      'password' = '1qaz!QAZ',
      'dbname' = 'test_cdc'
    );

  • 參數配置

    參數	說明
    type	類型，固定值為starrocks。
    endpoint	FE的內網地址和查詢連接埠，格式為EMR Serverless StarRocks執行個體FE節點的內網地址:9030。 例如，fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030。 說明 關於如何擷取EMR Serverless StarRocks執行個體FE節點的內網地址，請參見查看執行個體列表與詳情。
    username	StarRocks的使用者名稱。 填寫步驟一：準備測試資料中帳號的使用者名稱。本樣本為test。
    password	StarRocks資料庫服務的密碼。 填寫步驟一：準備測試資料中帳號的密碼。
    dbname	StarRocks資料庫名稱。 填寫步驟一：準備測試資料中建立的資料庫名。本樣本為test_cdc。

步驟三：建立並上線作業

1. 在阿里雲Realtime ComputeFlink控制台的資料開發 > ETL頁面，編寫CTAS語句。

   以下是三種樣本：

   • AtLeast once語義：通過sink.buffer-flush.interval-ms配置項，配置每次寫入StarRocks的時間間隔，優點是寫入間隔時間短，佔用記憶體較少。

     /*
           AtLeast once 語義
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
     'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
     'table-name'='runoob_tbl',
     'username'='test',
     'password' = '1qaz!QAZ',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl;
                                     

   • Exactly once語義：需要定義checkpoint間隔，優點是在各種異常情況下保障資料不丟失不重複，缺點是資料可見時間取決於checkpoint間隔。更多資訊，請參見Checkpointing。

     /*
           Exactly once 語義。
     */
     set 'execution.checkpointing.interval' = '1 min';
     set 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
     set 'execution.checkpointing.timeout' = '10 min';
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
     'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
     'table-name'='runoob_tbl',
     'username'='test',
     'password' = '1qaz!QAZ',
     'sink.semantic' = 'exactly-once',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl;
                                     

   • Simple模式：優點是建立表時不需要關注原表有哪些欄位，會按照MySQL的表格式照搬過來，開發人員使用比較方便。缺點是不能建立分區，對於需要分區的表，仍需要通過normal模式建立。

     /*
           上面兩個為normal模式，本樣本示範simple模式
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl with (
     'starrocks.create.table.mode'='simple',
      'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
     'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
     'table-name'='runoob_tbl',
     'username'='test',
     'password' = '1qaz!QAZ',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl;
                                     

   表 1. WITH參數

   參數	是否必選	描述
   starrocks.create.table.properties	是	StarRocks建表語句中除了欄位定義以外的其他尾碼定義，例如樣本中的engine、key和buckets等。
   database-name	是	StarRocks資料庫名稱。 本樣本為test_cdc。
   jdbc-url	是	用於在StarRocks中執行查詢操作。 例如，jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030。其中，fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com為EMR Serverless StarRocks執行個體FE節點的內網地址。 說明 關於如何擷取EMR Serverless StarRocks執行個體FE節點的內網地址，請參見查看執行個體列表與詳情。
   load-url	是	指定FE的內網地址和查詢連接埠，格式為EMR Serverless StarRocks執行個體FE節點的內網地址:8030。 例如，fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030。 說明 關於如何擷取EMR Serverless StarRocks執行個體FE節點的內網地址，請參見查看執行個體列表與詳情。
   sink.semantic	否	填寫exactly-once可以保障資料一致性語義，預設為at-least-once。
   starrocks.create.table.mode	否	支援以下參數值： normal模式（預設值）：必須像樣本一樣在starrocks.create.table.properties配置中填寫engine、key和buckets等完整的配置。 simple模式：預設選擇engine為olap，選擇key類型為primary key，且主鍵與MySQL的主鍵保持完全一致，預設distributed by hash（所有的主鍵），預設無分區。需要在starrocks.create.table.properties配置中填寫的必填內容為buckets ，選填內容為properties等配置。
   sink.properties.row_delimiter	否	自訂行分隔字元。
   sink.properties.column_separator	否	自訂欄分隔字元。

   說明

   • 因為vvr-6.0.5-flink-1.15及以上版本移除了sink.use.new-api，所以使用vvr-6.0.5-flink-1.15之前的版本時，請在with參數中添加'sink.use.new-api'='false',。

   • 其他配置請參見Continuously load data from Apache Flink。

   表 2. OPTIONS參數

   參數	描述
   connector	類型，固定值為mysql-cdc。
   hostname	RDS的內網地址。 您可以在RDS的資料庫連接頁面，單擊內網地址進行複製。例如，rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com。
   port	MySQL資料庫服務的連接埠號碼，預設值為3306。
   username	MySQL資料庫服務的使用者名稱。 填寫步驟一：準備測試資料中帳號的使用者名稱。本樣本為test。
   password	MySQL資料庫服務的密碼。 填寫步驟一：準備測試資料中帳號的密碼。
   table-name	StarRocks中的表名稱。 填寫步驟一：準備測試資料中建立的表名。本樣本為runoob_tbl。
   database-name	預設的MySQL資料庫名稱。 填寫步驟一：準備測試資料中建立的資料庫名。本樣本為test_cdc。

2. 單擊部署。

3. 在部署新版本頁面選擇好部署目標，單擊確定。

4. 在作業營運頁面，單擊目標作業操作列的啟動。

步驟四：驗證資料同步結果

CDAS介紹

CDAS是CTAS的一個文法糖。通過CDAS語句，可以實現MySQL中的整庫同步，即產生一個Flink Job，源表是MySQL中的Database，目標表是StarRocks中對應的多張表，同時可以使用including table文法，只選擇一個Database中的部分表進行CDAS操作。

CREATE DATABASE IF NOT EXISTS sr_db with (
'starrocks.create.table.properties'=' buckets 8',
'starrocks.create.table.mode'='simple',
'jdbc-url'='jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
'sink.buffer-flush.interval-ms' = '5000',
'sink.properties.row_delimiter' = '\x02',
'sink.properties.column_separator' = '\x01'
)
 as DATABASE mysql.test_cdc including table
 'tabl1','tbl2','tbl3';

相關文檔

Realtime ComputeFlink不僅支援CTAS語句同步資料至StarRocks，還支援通過資料攝入YAML方式同步資料至StarRocks。詳情請參見資料攝入。