流计算 Oceanus Pipeline

一、快速示例

source: type: mysql name: MySQL Source hostname: 127.0.0.1 port: 3306 username: admin password: pass tables: adb.\\.* server-id: 5401-5404sink: type: doris name: Doris Sink fenodes: 127.0.0.1:8030 username: root password: passpipeline: 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	模型自身的配置项，按所选模型的要求填写。