Flink CDC系列之:学习理解核心概念——Data Pipeline
- 数据管道
- source
- sink
- 管道配置
- Table ID
- route
- transform
- 案例
数据管道
由于 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