定义

由于在 Flink CDC 中，事件从上游流转到下游遵循 Pipeline 的模式，因此整个 ETL 作业也被称为 Data Pipeline。

参数

一个 pipeline 包含着 Flink 的一组算子链。 为了描述 Data Pipeline，我们需要定义以下部分：

• source
• sink
• pipeline

下面是 Data Pipeline 的一些可选配置：

• route
• transform

示例

只包含必需部分

我们可以使用以下 yaml 文件来定义一个简单的 Data Pipeline 来同步 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: ""

   pipeline:
     name: Sync MySQL Database to Doris
     parallelism: 2
     flink-conf:
       execution.checkpointing.interval: 2min

包含可选部分

我们可以使用以下 yaml 文件来定义一个复杂的 Data Pipeline 来同步 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

Pipeline 配置

下面是 Data Pipeline 级别支持的配置选项。 请注意，虽然这些参数都是可选的，但至少需要指定其中一个。也就是说，pipeline 部分是必需的，不能为空。

参数	含义	optional/required
name	这个 pipeline 的名称，会用在 Flink 集群中作为作业的名称。	optional
parallelism	pipeline 的全局并发度，默认值是 1。	optional
local-time-zone	作业级别的本地时区。	optional
execution.runtime-mode	pipeline 的运行模式，包含 STREAMING 和 BATCH，默认值是 STREAMING。	optional
route-mode	路由规则的匹配模式。可选值：ALL_MATCH（默认，应用所有匹配的规则）或 FIRST_MATCH（只应用第一个匹配的规则）。	optional
schema.change.behavior	如何处理 schema 变更。可选值：exception、evolve、try_evolve、lenient（默认值）或 ignore。	optional
schema.operator.uid	Schema 算子的唯一 ID。此 ID 用于算子间通信，必须在所有算子中保持唯一。已废弃：请使用 operator.uid.prefix 代替。	optional
schema-operator.rpc-timeout	SchemaOperator 等待下游 SchemaChangeEvent 应用完成的超时时间，默认值是 3 分钟。	optional
operator.uid.prefix	Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。	optional
sink.partitioning.strategy	Sink 写入数据时使用的分区策略。数据类型：String。默认值：SINK_DEFINED。备注：可配置的值如下：SINK_DEFINED：使用 Sink 定义的分区策略；PRIMARY_KEY：按表 ID 和主键分区；TABLE_ID：仅按表 ID 分区。	optional
transform.decimal.precision.mode	transform 表达式求值中 DECIMAL 类型的最大精度模式。可选值：UP_TO_19（默认，使用 Calcite 默认类型系统）或 UP_TO_38（允许 DECIMAL 精度最高为 38 位）。	optional
transform.async-execution.enabled	是否为 PostTransform 开启有序异步执行，默认值为 false。	optional
transform.async-execution.timeout	有序异步执行中每个 PostTransform 事件的超时时间，默认值为 5 分钟。	optional
transform.async-execution.capacity	异步执行中最多允许同时处理的 PostTransform 事件数，默认值为 100。	optional
transform.async-execution.worker-threads	每个异步 PostTransform 任务使用的工作线程数，默认值为 16。	optional
flink-conf	用于配置 Flink 相关参数。 配置优先级：命令行参数（CLI arguments）高于 pipeline.yaml 高于 Flink 默认配置 config.yaml。同时支持扁平键值对与嵌套 YAML 结构。	optional

实验性且不稳定： 异步 PostTransform 执行目前是实验性功能，其配置和行为可能会在未来版本中发生变化。

异步 PostTransform 适用于 AI 模型调用等 I/O 密集型表达式。DataChangeEvent 可能并发调用 UDF 和 AI 模型客户端，但输出和所有 SchemaChangeEvent 仍保持有序，因此 UDF 和 AI 模型客户端实现必须是线程安全的。为保证 schema 状态一致，checkpoint 或 savepoint 前会等待尚未完成的异步请求，长时间运行的请求可能会延长 checkpoint 时间。savepoint 只支持在并行度不变时恢复，并且不能在从已有 savepoint 恢复时开启或关闭此选项。

注意：虽然上述参数都是可选的，但至少需要指定其中一个。pipeline 部分是必需的，不能为空。