流计算 Oceanus Pipeline

本文说明 YAML 作业中 pipeline 块支持的配置项。这些参数作用于整个作业，与具体的 Source、Sink 类型无关。Source 与 Sink 各自的参数请参考对应的连接器文档。
﻿
pipeline 块是作业定义中必需的部分，不能为空，但其中的每个参数都是可选的。
﻿
快速示例
source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\\.*
  server-id: 5401-5404
﻿
sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: pass
﻿
pipeline:
  name: Sync MySQL to Doris
  parallelism: 4
  local-time-zone: Asia/Shanghai
  schema.change.behavior: LENIENT
  operator.uid.prefix: cdc-sync
﻿
配置项
参数
是否必填
默认值
类型
说明
name
optional
Flink CDC Pipeline Job
String
作业名称，会作为 Flink 集群中的作业名显示。
parallelism
optional
1
Integer
作业的全局并发度。
local-time-zone
optional
systemDefault
String
作业级别的本地时区，用于时间类型字段的转换。
默认值 systemDefault 表示使用作业运行环境的系统时区， 跨时区部署时结果可能不一致，建议显式配置。
取值可以是完整时区名（如 Asia/Shanghai、America/Los_Angeles）， 也可以是自定义时区 ID（如 GMT+08:00）。
execution.runtime-mode
optional
STREAMING
String
作业的运行模式，可取 STREAMING 或 BATCH。
设为 BATCH 时按批语义执行，可用于一次性全量同步。
注意：部分 Source 在 BATCH 模式下对启动模式有限制， 例如 MySQL 只允许 snapshot。
schema.change.behavior
optional
LENIENT
String
表结构变更的同步策略，可取 IGNORE、LENIENT、 TRY_EVOLVE、EVOLVE、EXCEPTION，取值大小写不敏感。 详见「表结构变更策略」章节。
schema.change.skip-duplicate-column.enabled
optional
false
Boolean
当目标表结构中已存在同名列时，是否跳过该新增列事件。
默认为 false，即保留历史行为，遇到重复列时 schema 校验失败。
schema.change.skip-create-table.enabled
optional
false
Boolean
是否跳过向下游执行建表，同时保留该事件用于内部表结构跟踪。
仅当目标表已提前创建且与作业的表结构兼容时才可开启。
route-mode
optional
ALL_MATCH
String
路由规则的匹配模式。
ALL_MATCH：应用所有匹配的路由规则。
FIRST_MATCH：只应用第一条匹配的规则并停止后续匹配。
operator.uid.prefix
optional
(none)
String
作业中所有算子 UID 的统一前缀。
未设置时由 Flink 自动生成 UID。建议显式设置，以便算子 UID 稳定可识别， 便于有状态升级、问题排查和 Flink UI 诊断。
schema.operator.uid
optional
$$_schema_operator_$$
String
表结构算子的唯一 ID，用于算子间通信，必须在所有算子中保持唯一。
已废弃，请使用 operator.uid.prefix 代替。
注意：该参数与 operator.uid.prefix 不能同时指定。
schema-operator.rpc-timeout
optional
3min
Duration
表结构算子等待下游表结构变更应用完成的超时时间。
下游（如 Doris、StarRocks）执行 DDL 较慢时，可适当调大该值。
﻿
LENIENT 模式的事件重写
LENIENT 是默认策略，它会在应用前重写部分事件，以避免下游丢数据：
﻿
新增列：新增的列统一转为可空（nullable）后添加，避免下游存在历史数据时新增非空列失败。
删除列：不真正删除该列，只是把该列转为可空；如果该列原本就是可空，则不做任何操作。
重命名列：不真正重命名，而是添加一个可空的新列（新列名），必要时再把旧列转为可空。
其它变更（如修改列类型）：原样应用。
﻿
此外，LENIENT 模式下默认不会把 drop.table（删除表）和 truncate.table（清空表）事件发送到下游，以避免误删数据。其它策略下默认不排除任何事件。该默认行为仅在完全没有配置 exclude.schema.changes 时生效。
﻿
按事件类型精细控制
如果需要按事件类型分别控制，可以在 sink 块配置 include.schema.changes 与exclude.schema.changes，后者优先级更高。可配置的事件类型如下：
事件类型
说明
create.table
创建表。
add.column
新增列。
alter.column.type
修改列类型。
drop.column
删除列。
rename.column
重命名列。
drop.table
删除表。
truncate.table
清空表数据。
支持部分匹配，例如传入 drop 等价于同时传入 drop.column 与 drop.table。
﻿
建表事件是所有后续表结构变更的基础。显式指定 include.schema.changes 时，create.table 会被自动加入，除非通过 
exclude.schema.changes 明确排除它。
﻿
示例：
sink:
  include.schema.changes: [create.table, column]   # 包含建表与所有列相关事件
  exclude.schema.changes: [drop.column]            # 但排除删除列
﻿
子配置块
pipeline 块下除了上述参数，还支持两个子配置块。
user-defined-function
声明自定义函数（UDF），供 transform 块中的 projection 与 filter 引用。
pipeline:
  user-defined-function:
    - name: addone
      classpath: com.example.functions.AddOneFunctionClass
    - name: format
      classpath: com.example.functions.FormatFunctionClass
      options:
        format: '%S'
字段
是否必填
说明
name
required
函数名，在 transform 表达式中按此名称引用。
classpath
required
函数实现类的全限定名。
options
optional
传给函数的配置项，以键值对形式声明。
model
声明内置模型，供 transform 块引用。模型是内置提供的函数，只需指定模型类名，无需写完整 classpath。
pipeline:
  model:
    - model-name: my-model
      class-name: BuiltInModelClass
      option-key: option-value
字段
是否必填
说明
model-name
required
模型名，在 transform 表达式中按此名称引用。
class-name
required
模型类名，取内置模型名即可。
其它字段
optional
模型自身的配置项，按所选模型的要求填写。

参数	是否必填	默认值	类型	说明
name	optional	Flink CDC Pipeline Job	String	作业名称，会作为 Flink 集群中的作业名显示。
parallelism	optional	1	Integer	作业的全局并发度。
local-time-zone	optional	systemDefault	String	作业级别的本地时区，用于时间类型字段的转换。默认值 systemDefault 表示使用作业运行环境的系统时区，跨时区部署时结果可能不一致，建议显式配置。取值可以是完整时区名（如 Asia/Shanghai、America/Los_Angeles），也可以是自定义时区 ID（如 GMT+08:00）。
execution.runtime-mode	optional	STREAMING	String	作业的运行模式，可取 STREAMING 或 BATCH。设为 BATCH 时按批语义执行，可用于一次性全量同步。注意：部分 Source 在 BATCH 模式下对启动模式有限制，例如 MySQL 只允许 snapshot。
schema.change.behavior	optional	LENIENT	String	表结构变更的同步策略，可取 IGNORE、LENIENT、 TRY_EVOLVE、EVOLVE、EXCEPTION，取值大小写不敏感。详见「表结构变更策略」章节。
schema.change.skip-duplicate-column.enabled	optional	false	Boolean	当目标表结构中已存在同名列时，是否跳过该新增列事件。默认为 false，即保留历史行为，遇到重复列时 schema 校验失败。
schema.change.skip-create-table.enabled	optional	false	Boolean	是否跳过向下游执行建表，同时保留该事件用于内部表结构跟踪。仅当目标表已提前创建且与作业的表结构兼容时才可开启。
route-mode	optional	ALL_MATCH	String	路由规则的匹配模式。 ALL_MATCH：应用所有匹配的路由规则。 FIRST_MATCH：只应用第一条匹配的规则并停止后续匹配。
operator.uid.prefix	optional	(none)	String	作业中所有算子 UID 的统一前缀。未设置时由 Flink 自动生成 UID。建议显式设置，以便算子 UID 稳定可识别，便于有状态升级、问题排查和 Flink UI 诊断。
schema.operator.uid	optional	$$_schema_operator_$$	String	表结构算子的唯一 ID，用于算子间通信，必须在所有算子中保持唯一。已废弃，请使用 operator.uid.prefix 代替。注意：该参数与 operator.uid.prefix 不能同时指定。
schema-operator.rpc-timeout	optional	3min	Duration	表结构算子等待下游表结构变更应用完成的超时时间。下游（如 Doris、StarRocks）执行 DDL 较慢时，可适当调大该值。

事件类型	说明
create.table	创建表。
add.column	新增列。
alter.column.type	修改列类型。
drop.column	删除列。
rename.column	重命名列。
drop.table	删除表。
truncate.table	清空表数据。

字段	是否必填	说明
name	required	函数名，在 transform 表达式中按此名称引用。
classpath	required	函数实现类的全限定名。
options	optional	传给函数的配置项，以键值对形式声明。

Pipeline

本页目录：

快速示例

配置项

LENIENT 模式的事件重写

按事件类型精细控制

子配置块

user-defined-function

model