Realtime Compute for Apache Flink：Flink CDC Source和Sink模組

文法

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  
sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog

您可以在 source和 sink模組中使用using.built-in-catalog文法引用已建立的內建Catalog。

例如，在上面的例子中，mysql_rds_catalog的Catalog中繼資料中已經包含hostname、username、password等必填參數，因此無需在YAML作業中重複提供這些參數。

使用限制

下述連接器已經支援複用Catalog的串連資訊：

• MySQL（源端）

• Kafka（源端）

• Upsert Kafka（目標端）

• StarRocks（目標端）

• Hologres（目標端）

• Paimon（目標端）

• SLS（源端）

• Iceberg（目標端）

文法

source:                                                                                                                                                                                                                                                                                                          
  type: mysql                                                                                                                                                                                                                                                                                                     
  host: localhost                                                                                                                                                                                                                                                                                                  
  port: 3306                                                                                                                                                                                                                                                                                                       
  username: admin
  password: pass
  tables: mydb.orders
  source-expand:
  # 將 mydb.orders 複製為三張表，分別經過不同處理後寫入下遊
    - input-table: mydb.orders
      output-table: [ dwd.orders_full, dws.orders_summary, ads.orders_report ]

# 對三張擴充後的表分別應用不同的 transform
transform:
  # 明細層：保留全部欄位，增加計算資料行
  - source-table: dwd.orders_full
    projection: "*, amount * discount as final_price"
  # 匯總層：只保留關鍵字段，過濾小額訂單
  - source-table: dws.orders_summary
    projection: order_id, user_id, amount, order_status
    filter: amount > 100
    primary-keys: order_id
  # 報表層：只保留統計需要的欄位，過濾已取消的訂單
  - source-table: ads.orders_report
    projection: order_id, user_id, amount, TO_UPPER(order_status) as status
    filter: order_status <> 'CANCELLED'
    primary-keys: order_id

# 將三張擴充後的表路由到不同的下遊目標表
route:
  - source-table: dwd.orders_full
    sink-table: starrocks_dwd.orders_full_detail
  - source-table: dws.orders_summary
    sink-table: starrocks_dws.orders_summary
  - source-table: ads.orders_report
    sink-table: starrocks_ads.orders_report

sink:
  type: starrocks
  name: sink-starrocks
  jdbc-url: jdbc:mysql://localhost:9030
  load-url: localhost:8030
  username: root
  password: pass

使用如上所示格式進行配置，例子中mydb.orders 表分發複製為 dwd.orders_full， dws.orders_summary和ads.orders_report三張表，每個表經過不同的處理邏輯後分別寫入到下遊的三張表中。

注意事項

• VVR 11.6 及以上版本可用。

• 資料分發後預設不保留原有表。以文法部分中例子為例，mydb.orders表按如上配置後，下發的資料中不再包含mydb.orders表。如果想保留mydb.orders表，需要將mydb.orders表添加到output-table 部分。

  source-expand:
    - input-table: mydb.orders
      output-table: [ mydb.orders, db1.orders, db2.orders ]

• input-table 和 output-table 中不支援使用Regex。