数据管道

由于 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 Dorisparallelism: 2user-defined-function:- name: addoneclasspath: com.example.functions.AddOneFunctionClass- name: formatclasspath: 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: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*sink:type: dorisfenodes: 127.0.0.1:8030username: rootpassword: ""transform:- source-table: adb.web_order01projection: \*, UPPER(product_name) as product_namefilter: id > 10 AND order_id > 100description: project fields and filter- source-table: adb.web_order02projection: \*, UPPER(product_name) as product_namefilter: id > 20 AND order_id > 200description: project fields and filterroute:- source-table: app_db.orderssink-table: ods_db.ods_orders- source-table: app_db.shipmentssink-table: ods_db.ods_shipments- source-table: app_db.productssink-table: ods_db.ods_productspipeline:name: Sync MySQL Database to Dorisparallelism: 2

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

 source:type: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*sink:type: dorisfenodes: 127.0.0.1:8030username: rootpassword: ""transform:- source-table: adb.web_order01projection: \*, format('%S', product_name) as product_namefilter: addone(id) > 10 AND order_id > 100description: project fields and filter- source-table: adb.web_order02projection: \*, format('%S', product_name) as product_namefilter: addone(id) > 20 AND order_id > 200description: project fields and filterroute:- source-table: app_db.orderssink-table: ods_db.ods_orders- source-table: app_db.shipmentssink-table: ods_db.ods_shipments- source-table: app_db.productssink-table: ods_db.ods_productspipeline:name: Sync MySQL Database to Dorisparallelism: 2user-defined-function:- name: addoneclasspath: com.example.functions.AddOneFunctionClass- name: formatclasspath: com.example.functions.FormatFunctionClass