Fluss Pipeline 连接器

Fluss Pipeline 连接器可用作 Pipeline 的 Data Source 或 Data Sink，从 Fluss 读取或向其写入数据。本文档介绍这两种用法的配置。

What can the connector do?

• 自动创建不存在的表
• 数据同步
• Schema 变更同步（lenient 模式）
• 动态 Source 表订阅

Fluss Source

以下是动态发现 Fluss 表并读取的最小配置：

source:
  type: fluss
  bootstrap.servers: localhost:9123
  table.discoverer.type: fluss-default
  table.discoverer.pattern: 'inventory\..*'
  scan.discovery.interval: 10 s
  scan.startup.mode: earliest

table.discoverer.type 用于选择 Source 的表发现器。fluss-default 通过 table.discoverer.pattern 匹配全限定表名；选择其他发现器时，需配置其必需的 table.discoverer.* 参数。

How to create Pipeline

从 MySQL 读取数据并写入 Fluss 的 Pipeline 可定义如下：

source:
   type: mysql
   name: MySQL Source
   hostname: 127.0.0.1
   port: 3306
   username: admin
   password: pass
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: fluss
  name: Fluss Sink
  bootstrap.servers: localhost:9123
  # Security-related properties for the Fluss client
  properties.client.security.protocol: sasl
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: developer
  properties.client.security.sasl.password: developer-pass

pipeline:
  name: MySQL to Fluss Pipeline
  parallelism: 2
  schema.change.behavior: LENIENT

Pipeline Connector Options

Option	Required	Default	Type	Description
type	required	(none)	String	指定要使用的连接器，这里需要设置成 'fluss'。
name	optional	(none)	String	Sink 的名称。
bootstrap.servers	required	(none)	String	用于建立与 Fluss 集群初始连接的主机/端口对列表。
sink.partitioning.strategy	optional	DEFAULT	String	DataChangeEvent 路由使用的分区策略。可选值为 DEFAULT 和 FORWARD。DEFAULT 对主键表按主键进行哈希分区，对日志表采用随机分发，将事件分散到各下游 subtask 以实现负载均衡。FORWARD 将数据事件发送到与上游 subtask 索引相同的下游 subtask，用于 Fluss 到 Fluss 的数据同步。上游数据分布必须与 Fluss 表的分桶策略保持一致，否则可能出现数据正确性问题。
bucket.key	optional	(none)	String	指定每个 Fluss 表的数据分布策略。表之间用 ';' 分隔，分桶键之间用 ',' 分隔。格式：database1.table1:key1,key2;database1.table2:key3。 数据将根据分桶键的哈希值分配到各个桶中（分桶键必须是主键的子集，且不包含主键表的分区键）。 若表有主键但未指定分桶键，则分桶键默认为主键（不含分区键）；若表无主键且未指定分桶键，则数据将随机分配到各个桶中。
bucket.num	optional	(none)	String	每个 Fluss 表的桶数量。表之间用 ';' 分隔。格式：database1.table1:4;database1.table2:8。
properties.table.*	optional	(none)	String	将 Fluss table 支持的参数传递给 pipeline，参考，See Fluss table options。
properties.client.*	optional	(none)	String	将 Fluss client 支持的参数传递给 pipeline，See Fluss client options。

在 transform 规则中配置表参数

可以通过 transform 规则中的 table-options 为匹配的表配置 Fluss 表参数。 参数名直接使用 table.replication.factor 等 Fluss 表属性名称，无需添加 sink 配置中的 properties. 前缀。 这些参数会覆盖同名的 sink.properties.table.* 配置，未覆盖的 sink 表参数仍然生效。

transform:
  - source-table: inventory.products
    projection: "*"
    table-options: table.replication.factor=2,table.log.ttl=7d

sink:
  type: fluss
  bootstrap.servers: localhost:9123
  properties.table.replication.factor: 3

在此示例中，为 inventory.products 创建的 Fluss 表副本数为 2，日志保留时间为 7 天，其他表的副本数为 3。 这些参数仅在建表时生效，不会修改已有表的属性。 分桶键和桶数量仍通过 sink 的 bucket.key 和 bucket.num 配置。

使用说明

• 支持 Fluss 主键表和日志表。

动态 Source 订阅

当 Fluss 作为带表发现器的 Source 使用时，每次成功发现的结果都是当前订阅表的完整权威集合。配置正数 scan.discovery.interval 才会周期性更新订阅。空结果会退订全部已发现的表；发现失败不会修改当前订阅，并会使作业失败。

退订表只会停止并清理 Source 侧 reader，不会删除 Fluss 表，也不会改变 Sink 行为。恢复后的 reader 会先等待新的订阅快照， 再打开 checkpoint 中恢复的 split，因此当前仍处于退订状态的表不会通过恢复的 split 输出记录。表再次被订阅时会作为新表处理， 并使用配置的 scan.startup.mode。

移除与 checkpoint 状态协同：故障恢复时会从最近一次成功 checkpoint 恢复 Source split 和待移除 tombstone；订阅由发现流程刷新。如果退订和重新订阅都发生在 相邻两次已完成的 checkpoint 之间，故障回滚时可以表现为从未退订过；若人为恢复到移除 tombstone 之前的 checkpoint， 而该表当前已重新订阅，则不承诺重新开始一个全新的表生命周期。对于主键表，移除时不会提前释放快照 lease，仍由现有的 过期和关闭逻辑处理。

• 关于自动建表

  • 没有分区键
  • 桶数量由 bucket.num 选项控制
  • 数据分布由 bucket.key 选项控制。对于主键表，若未指定分桶键，则分桶键默认为主键（不含分区键）；对于无主键的日志表，若未指定分桶键，则数据将随机分配到各个桶中。

• 支持在 lenient 模式下进行 Schema 变更同步，通过 schema.change.behavior: lenient 配置。支持以下 Schema 变更事件：

  • 新增列 — 新列会追加到 Fluss 表中。
  • 删除列 — 在 lenient 模式下不会真正删除列，而是忽略该删除操作，后续写入时将该列的值设为 null。
  • 重命名列 — 在 lenient 模式下，此操作会被转换为新增列 + 将旧列类型修改为可空的序列。
  • 修改列类型 — 不支持。

  要启用 Schema 变更同步，请在 pipeline 中配置 schema.change.behavior: lenient。如果想要忽略所有 Schema 变更，使用 schema.change.behavior: IGNORE。

• 关于数据同步，Pipeline 连接器使用 Fluss Java Client 向 Fluss 写入数据。

Data Type Mapping

Flink CDC type	Fluss type	Note
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
DATE	DATE
TIME	TIME
TIMESTAMP	TIMESTAMP
TIMESTAMP_LTZ	TIMESTAMP_LTZ
CHAR(n)	CHAR(n)
VARCHAR(n)	STRING
BINARY(n)	BINARY(n)
VARBINARY(N)	BYTES
ARRAY	ARRAY	元素类型递归映射。
MAP	MAP	键和值类型递归映射。
ROW	ROW	字段类型递归映射。