Realtime Compute for Apache Flink：Flink CDC資料攝入作業開發（公測中）

單擊目標工作空間操作列的控制台。

在左側導覽列選擇資料開發 > 資料攝入。

單擊後，單擊從模板建立。

選擇目標資料同步模板。

目前僅支援MySQL到StarRocks、MySQL到Paimon和MySQL到Hologres這三個模板。

填寫作業資訊（作業名稱、儲存位置和引擎版本）後，單擊確定。

配置Flink CDC作業Source和Sink相關資訊。

參數配置資訊可查看相關連接器文檔。

單擊目標工作空間操作列的控制台。

在左側導覽列選擇資料開發 > 資料攝入。

單擊，選擇建立資料攝入草稿，填寫檔案名稱和引擎版本，單擊建立。

複製開源社區的Flink CDC作業。

（可選）單擊深度檢查。

您可以進行文法檢測、網路連通性和存取權限檢查。

單擊目標工作空間操作列的控制台。

在左側導覽列選擇資料開發 > 資料攝入。

單擊，選擇建立資料攝入草稿，填寫檔案名稱和引擎版本，單擊建立。

配置Flink CDC作業開發資訊。

# 必填
source:
  # 資料來源類型
  type: <替換為您源端連接器類型>
  # 資料來源配置資訊，配置項詳情請參見對應連接器文檔。
  ...

# 必填
sink:
  # 目標類型
  type: <替換為您目標端連接器類型>
  # 資料目標配置資訊，配置項詳情請參見對應連接器文檔。
  ...

# 可選
transform:
  # 轉換規則，針對flink_test.customers表
  - source-table: flink_test.customers
    # 投影配置，指定要同步的列，並進行資料轉換
    projection: id, username, UPPER(username) as username1, age, (age + 1) as age1, test_col1, __schema_name__ || '.' || __table_name__ identifier_name
    # 過濾條件，只同步id大於10的資料
    filter: id > 10
    # 描述資訊，用於解釋轉換規則
    description: append calculated columns based on source table

# 可選
route:
  # 路由規則，指定源表和目標表之間的對應關係
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # 描述資訊，用於解釋路由規則
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # 描述資訊，用於解釋路由規則
    description: sync customers_suffix table

#可選
pipeline:
  # 任務名稱
  name: MySQL to Hologres Pipeline

涉及的代碼塊說明詳情如下。

各模組文法結構和配置項說明詳情，請參見Flink CDC資料攝入作業開發參考。

以將MySQL中app_db資料庫下的所有表同步到Hologres的某個資料庫為例，程式碼範例如下。

source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: Sync MySQL Database to Hologres

（可選）單擊深度檢查。

您可以進行文法檢測、網路連通性和存取權限檢查。