流计算连接器 SQL 使用说明

作为 Source

配置前需确认以下约束：

• topic 与 topic-pattern 二选一，不可同时配置。

• scan.startup.mode 为 specific-offsets 时需指定 scan.startup.specific-offsets，为 timestamp 时需指定 scan.startup.timestamp-millis。

• 使用 key.format 时需同时指定 key.fields。

CREATE TABLE kafka_source (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  -- 元数据字段（可选）
  `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
  `partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,
  `offset` BIGINT METADATA FROM 'offset' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'your_topic',
  'properties.bootstrap.servers' = 'alikafka-xxx:9092',
  'properties.group.id' = 'your_consumer_group',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'json'
);

Source 参数说明：

作为 Sink

sink.delivery-guarantee 为 exactly-once 时需配合事务，并通过 sink.transactional-id-prefix 指定唯一的事务 ID 前缀。

CREATE TABLE kafka_sink (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'your_sink_topic',
  'properties.bootstrap.servers' = 'alikafka-xxx:9092',
  'format' = 'json',
  'sink.partitioner' = 'default'
);

Sink 参数说明：

数据格式（Format 常用参数）

Kafka Source 和 Sink 均通过 format 或 value.format 指定消息体的数据格式：Source 将消息体反序列化为表中的字段，Sink 将表中的字段序列化为消息体。常用取值包括 json、csv、avro、raw、debezium-json 和 canal-json。不同格式有各自的附加参数，按 格式名.参数 的形式配置到 WITH 子句，例如 json.ignore-parse-errors。

JSON 格式通过 'format' = 'json' 或 'value.format' = 'json' 指定，附加参数如下：

CSV 格式通过 'format' = 'csv' 或 'value.format' = 'csv' 指定，附加参数如下：

以下格式均支持 Source 和 Sink：

• raw：原始消息体格式，适合单字段读取和写入。

• avro：Avro 二进制格式。

• debezium-json/canal-json：用于读取或输出 CDC 变更消息。

作为 Source

配置前需确认以下约束：

• source.type 决定必填项：topic 模式需指定 topic 且禁用 group-id，status 模式需指定 group-id 且禁用 topic。

• Source 的 topic 仅支持一个一级 Topic，不支持多级路径及 +、# 通配符。

• 未配置 consumer-group 时，source.commit-offsets-on-checkpoint 不能设为 true；scan.startup.mode 与 scan.new-partition.startup.mode 取 committed 也需先配置 consumer-group。

• source.empty-pull-backoff.max 不能小于 source.empty-pull-backoff.initial。

CREATE TABLE mqtt_source (
  `user_id` STRING,
  `user_name` STRING,
  `action` STRING,
  `event_time` TIMESTAMP(3),
  -- 元数据字段（可选）
  `mqtt_client_id` STRING METADATA VIRTUAL,
  `mqtt_qos_level` STRING METADATA VIRTUAL,
  `mqtt_real_topic` STRING METADATA VIRTUAL
) WITH (
  'connector' = 'mqtt',
  'topic' = 'stream-source-1',
  'instance-id' = 'mqtt-cn-xxxxxxxxx',
  'endpoints' = 'mqtt-cn-xxxxxxxxx-single-internal-vpc.mqtt.aliyuncs.com:5673',
  'access-key' = '${secret_ak}',
  'secret-key' = '${secret_sk}',
  'scan.startup.mode' = 'latest',
  'source.pull.batch-size' = '32',
  'source.pull.interval' = '100 ms',
  'source.pull.suspend-timeout' = '30 s',
  'message.field-delimiter' = ';'
);

Source 参数说明：

MQTT 可用元数据列：

Source 使用 METADATA FROM '<元数据名称>' VIRTUAL；列名与元数据名称一致时可省略 FROM。元数据不参与消息体编解码。

作为 Sink

topic 需为完整的 MQTT 发布 Topic，不支持 +、# 通配符或空层级（如 /a、a/、a//b）。

CREATE TABLE mqtt_sink (
  `user_id` STRING,
  `user_name` STRING,
  `action` STRING
) WITH (
  'connector' = 'mqtt',
  'topic' = 'stream-sink-1/C',
  'instance-id' = 'mqtt-cn-xxxxxxxxx',
  'endpoints' = 'mqtt-cn-xxxxxxxxx-single-internal-vpc.mqtt.aliyuncs.com:5673',
  'access-key' = '${secret_ak}',
  'secret-key' = '${secret_sk}',
  'sink.delivery-guarantee' = 'at-least-once',
  'message.field-delimiter' = ';'
);

Sink 参数说明：

作为 Source

rocketmq.source.group 需在控制台预先创建。

CREATE TABLE rocketmq_source (
  `user_id` STRING,
  `user_name` STRING,
  `action` STRING,
  `event_time` TIMESTAMP(3),
  `topic` STRING METADATA VIRTUAL,
  `msg_id` STRING METADATA VIRTUAL,
  `store_timestamp` BIGINT METADATA VIRTUAL,
  `born_timestamp` BIGINT METADATA VIRTUAL,
  `queue_id` INT METADATA VIRTUAL,
  `queue_offset` BIGINT METADATA VIRTUAL,
  `keys` STRING METADATA VIRTUAL,
  `tags` STRING METADATA VIRTUAL
) WITH (
  'connector' = 'rocketmq',
  'rocketmq.source.topic' = 'source',
  'rocketmq.source.group' = 'GID-flink',
  'rocketmq.client.endpoints' = 'rmq-cn-xxx-vpc.cn-hangzhou.rmq.aliyuncs.com:8080',
  -- 4.x 独享实例需带 namespace（实例 ID）：
  -- 'rocketmq.client.namespace' = 'MQ_INST_xxxxxxxx',
  'rocketmq.client.accessKey' = '${secret_ak}',
  'rocketmq.client.secretKey' = '${secret_sk}',
  'rocketmq.source.filter.tag' = '*',
  'rocketmq.source.startup.scan.mode' = 'latest',
  'rocketmq.source.pull.batch.size' = '32',
  'rocketmq.source.pull.interval.ms' = '100',
  'rocketmq.source.pull.suspend.timeout' = '30000',
  'rocketmq.client.message.field.delimiter' = ';',
  'rocketmq.client.message.line.delimiter' = '\n',
  'rocketmq.client.message.encoding' = 'UTF-8',
  'rocketmq.client.message.length.check' = 'NONE',
  'rocketmq.client.timeZone' = 'Asia/Shanghai',
  'rocketmq.client.partition.discovery.interval.ms' = '10000'
);

连接与鉴权参数（rocketmq.client.*，Source/Sink 通用）：

消费端参数（rocketmq.source.*）：

作为 Sink

rocketmq.sink.delivery.guarantee 取 EXACTLY_ONCE 时，需实例与连接器版本支持事务。

CREATE TABLE rocketmq_sink (
  `user_id` STRING,
  `user_name` STRING,
  `action` STRING,
  `process_time` TIMESTAMP(3),
  `field_a` STRING,
  `field_b` STRING,
  `keys` STRING METADATA,
  `tags` STRING METADATA
) WITH (
  'connector' = 'rocketmq',
  'rocketmq.sink.topic' = 'sink',
  'rocketmq.sink.group' = 'PID-flink',
  'rocketmq.client.endpoints' = 'rmq-cn-xxx-vpc.cn-hangzhou.rmq.aliyuncs.com:8080',
  'rocketmq.client.accessKey' = '${secret_ak}',
  'rocketmq.client.secretKey' = '${secret_sk}',
  'rocketmq.sink.delivery.guarantee' = 'AT_LEAST_ONCE',
  'rocketmq.client.message.field.delimiter' = ';',
  'rocketmq.sink.key.columns' = 'field_a,user_name',
  'rocketmq.sink.key.columns.included' = 'true',
  'rocketmq.sink.tag.column' = 'field_b',
  'rocketmq.sink.tag.column.included' = 'false'
);

生产端参数（rocketmq.sink.*）：

RocketMQ 可用元数据列：