数据管道

由于 Flink CDC 中的事件以管道方式从上游流向下游，因此整个 ETL 任务被称为数据管道。

管道对应于 Flink 中的一系列操作。

要描述数据管道，需要以下部分：

• source
• sink
• pipeline

以下部分是可选的：

• route
• transform

source

数据源用于访问元数据并从外部系统读取更改的数据。

数据源可以同时从多个表读取数据。

• Flink CDC系列之：学习理解核心概念——Data Source

sink

数据接收器用于应用架构更改并将更改数据写入外部系统。

数据接收器可以同时写入多个表。

• Flink CDC系列之：学习理解核心概念——Data Sink

管道配置

支持以下数据管道级别的配置选项：

参数	含义	可选/必需
name	管道的名称，将作为作业名称提交给Flink集群。	可选
parallelism	管道的全局并行度。默认为 1。	可选
local-time-zone	本地时区定义当前会话时区id。	可选

   pipeline:
     name: Sync MySQL Database to Doris
     parallelism: 2
     user-defined-function:
       - name: addone
         classpath: com.example.functions.AddOneFunctionClass
       - name: format
         classpath: com.example.functions.FormatFunctionClass

Table ID

在连接外部系统时，需要与外部系统的存储对象建立映射关系，这就是 Table Id 所指的。

为了兼容大多数外部系统，Table Id 用三元组表示：(namespace, schemaName, tableName)。

连接器应该建立 Table Id 与外部系统中存储对象的映射。

下表列出了不同数据系统的 Table Id 中的部分：

route

Route 指定匹配一串 source-table 到 sink-table 的规则，最典型的场景是分库分表合并，将多个上游 source 表路由到同一张 sink 表。

• Flink CDC系列之：学习理解核心概念——Route

transform

Transform模块帮助用户根据表中的数据列进行数据列的删除和扩展。
此外，它还可以帮助用户在同步过程中过滤一些不必要的数据。

• Flink CDC系列之：学习理解核心概念——Transform

案例

我们可以使用以下 yaml 文件来定义一个简洁的数据管道，描述将 MySQL app_db 数据库下的所有表同步到 Doris：

 source:
     type: mysql
     hostname: localhost
     port: 3306
     username: root
     password: 123456
     tables: app_db.\.*

   sink:
     type: doris
     fenodes: 127.0.0.1:8030
     username: root
     password: ""

   transform:
     - source-table: adb.web_order01
       projection: \*, UPPER(product_name) as product_name
       filter: id > 10 AND order_id > 100
       description: project fields and filter
     - source-table: adb.web_order02
       projection: \*, UPPER(product_name) as product_name
       filter: id > 20 AND order_id > 200
       description: project fields and filter

   route:
     - source-table: app_db.orders
       sink-table: ods_db.ods_orders
     - source-table: app_db.shipments
       sink-table: ods_db.ods_shipments
     - source-table: app_db.products
       sink-table: ods_db.ods_products

   pipeline:
     name: Sync MySQL Database to Doris
     parallelism: 2

我们可以使用以下 yaml 文件来定义一个复杂的数据管道，描述将 MySQL app_db 数据库下的所有表同步到 Doris，并给出特定的目标数据库名称 ods_db 和特定的目标表名称前缀 ods_ ：

 source:
     type: mysql
     hostname: localhost
     port: 3306
     username: root
     password: 123456
     tables: app_db.\.*

   sink:
     type: doris
     fenodes: 127.0.0.1:8030
     username: root
     password: ""

   transform:
     - source-table: adb.web_order01
       projection: \*, format('%S', product_name) as product_name
       filter: addone(id) > 10 AND order_id > 100
       description: project fields and filter
     - source-table: adb.web_order02
       projection: \*, format('%S', product_name) as product_name
       filter: addone(id) > 20 AND order_id > 200
       description: project fields and filter

   route:
     - source-table: app_db.orders
       sink-table: ods_db.ods_orders
     - source-table: app_db.shipments
       sink-table: ods_db.ods_shipments
     - source-table: app_db.products
       sink-table: ods_db.ods_products

   pipeline:
     name: Sync MySQL Database to Doris
     parallelism: 2
     user-defined-function:
       - name: addone
         classpath: com.example.functions.AddOneFunctionClass
       - name: format
         classpath: com.example.functions.FormatFunctionClass