| 连接器 | 'connector' = 'kafka'（需对应版本 connector jar 放 lib） |
|---|---|
| 启动位点 | earliest-offset / latest-offset / group-offsets / timestamp |
| 格式 | 'format' = 'json' 等，需要 flink-json 支持 |
Kafka 源表定义
Kafka 源表定义
CREATE TABLE orders_src (
order_id STRING,
amount DECIMAL(10,2),
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092',
'properties.group.id' = 'orders_sql',
'scan.startup.mode' = 'group-offsets',
'format' = 'json'
);connector jar 与 Flink 大版本绑定：升级 Flink 时连接器要同步替换，否则提交时报找不到 connector。