Fluss数据摄入连接器-实时计算 Flink版(Flink)-阿里云帮助中心

语法结构

source:
   type: xxx

sink:
  type: fluss
  name: Fluss Sink
  bootstrap.servers: localhost:9123
  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:
  schema.change.behavior: IGNORE

表结构变更

Fluss Server支持追加新增列，不支持其他列类型操作（比如删除列、重命名列和变更列类型）。因此，建议搭配Flink CDC默认的schema.change.behavior = LENIENT使用：

列重命名时，改为发送更改列类型和新增列两个事件。原有的列不删除，更改列类型为nullable，同时新增一个列名为新名称、数据类型改为nullable的列。
删除列时，改为发送更改列类型事件，将对应字段类型变为nullable。
新增列时仍发送新增列事件，但字段类型会变为nullable。

说明

仅VVR 11.6及以上版本支持。

配置项

参数	说明	是否必填	数据类型	默认值	备注
type	sink类型。	是	STRING	无	固定值为fluss。
name	sink名称。	否	STRING	无	无。
bootstrap.servers	Fluss Server地址	是	STRING	无	格式为host:port,host:port,host:port，以英文逗号（,）分隔。
bucket.key	分桶键	否	String	无	指定每个 Fluss 表的数据分布策略。表之间用`';'`分隔，分桶键之间用`','`分隔。格式：`database1.table1:key1,key2;database1.table2:key3`。数据将根据分桶键的哈希值分配到各个桶中（分桶键必须是主键的子集，且不包含主键表的分区键）。若表有主键但未指定分桶键，则分桶键默认为主键（不含分区键）；若表无主键且未指定分桶键，则数据将随机分配到各个桶中。 bucket.key和bucket.num值在创建表后就不再改变。
bucket.num	配置 fluss 表的桶数量	否	String	无	每个 Fluss 表的桶数量。表之间用`';'`分隔。格式：`database1.table1:4;database1.table2:8`。如果一张表未配置桶数量，默认会采用Server端的配置项`default.bucket.number`。 bucket.key和bucket.num值在创建表后就不再改变。
properties.table.*	fluss 表属性	否	String	无	将 Fluss table 支持的参数传递给 pipeline，详情请参见Fluss table options。
properties.client.*	fluss client参数	否	String	无	将 Fluss client 支持的参数传递给 pipeline，详情请参见Fluss client options。

类型映射

数据摄入类型映射如下表所示。

CDC字段类型	Fluss 字段类型
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
ARRAY	ARRAY
MAP	MAP
ROW	ROW
BINARY(n)	暂不支持
VARBINARY(N)	暂不支持

使用示例

将示例中的连接地址、账号和密码替换为实际值，在实时计算开发控制台创建数据摄入作业，复制 YAML 后部署并启动。并行度在作业配置中设置。

MySQL 数据同步至 Fluss

按 MySQL数据摄入连接器完成源端准备，在 MySQL 中执行以下 SQL。请使用未占用的测试库名。

CREATE DATABASE fluss_demo;

CREATE TABLE fluss_demo.orders (
    order_id BIGINT NOT NULL,
    customer_id BIGINT NOT NULL,
    amount DECIMAL(12, 2) NOT NULL,
    status VARCHAR(16) NOT NULL,
    PRIMARY KEY (order_id)
);

INSERT INTO fluss_demo.orders VALUES
    (1001, 101, 100.50, 'CREATED'),
    (1002, 102, 80.25, 'PAID'),
    (1003, 101, 60.25, 'PAID'),
    (1004, 101, 39.25, 'PAID'),
    (1005, 102, 19.75, 'CREATED'),
    (1006, 103, 50.50, 'PAID');

以下作业将订单表的全量数据和后续 Binlog 变更同步至 Fluss，沿用源库名、表名和主键。server-id 范围需未被其他复制客户端占用，ID 数量不少于源端并行度。

source:
  type: mysql
  name: MySQL Orders Source
  hostname: <mysql-host>
  port: 3306
  username: <mysql-cdc-user>
  password: <mysql-cdc-password>
  tables: fluss_demo.orders
  server-id: 5401-5404
  # Binlog 解析成为瓶颈且 CPU 资源充足时，可取消下一行注释。
  # scan.parallel-deserialize-changelog.enabled: true
  # 大表全量阶段可在首次启动前启用以下实验性选项，优先读取无上界分片。
  # scan.incremental.snapshot.unbounded-chunk-first.enabled: true

sink:
  type: fluss
  name: Fluss Orders Sink
  bootstrap.servers: <fluss-bootstrap-host:port>
  bucket.num: fluss_demo.orders:4
  properties.client.security.protocol: SASL
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: <fluss-user>
  properties.client.security.sasl.password: <fluss-password>

pipeline:
  schema.change.behavior: LENIENT

作业启动后，Fluss 中将生成 fluss_demo.orders 表。修改 MySQL 中的订单金额或删除订单，变更将同步至对应 Fluss 表。

Kafka JSON 数据写入 Fluss 主键表

在 Kafka 中创建 users Topic，并写入以下两条 JSON 消息，每行一条消息。

{"id":1001,"name":"Alice","age":28}
{"id":1002,"name":"Bob","age":32}

以下作业从 Topic 最早位点读取数据，通过 transform 指定 id 为主键，通过 route 将目标表设置为 fluss_demo.users。示例中的 Kafka 未开启认证；如已开启，请按 Kafka数据摄入连接器配置认证参数。

source:
  type: kafka
  name: Kafka Users Source
  properties.bootstrap.servers: <kafka-bootstrap-host:port>
  topic: users
  scan.startup.mode: earliest-offset
  value.format: json

sink:
  type: fluss
  name: Fluss Users Sink
  bootstrap.servers: <fluss-bootstrap-host:port>
  properties.client.security.protocol: SASL
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: <fluss-user>
  properties.client.security.sasl.password: <fluss-password>

transform:
  - source-table: users
    projection: "*"
    primary-keys: id

route:
  - source-table: users
    sink-table: fluss_demo.users

pipeline:
  schema.change.behavior: LENIENT

继续向 users Topic 写入以下消息，Fluss 表中 id 为 1001 的记录将更新为 age = 29，id 为 1002 的记录保持不变。

{"id":1001,"name":"Alice","age":29}