用 Flink SQL 接 Kafka
在 Flink SQL 里把 Kafka 主题声明成一张表，就可以用标准 SQL 做过滤、聚合与写出：源表指定连接器类型、主题名、Broker 地址、启动位点与数据格式；结果表同样用 Kafka 连接器，指定要写出的主题。加工逻辑全在 SQL 里，不用写 Java 代码。
启动位点决定读到哪里
首次启动或位点丢失时，启动位点决定从哪开始读：earliest-offset 从最早可用位置开始，会把历史全部重放；latest-offset 只消费新数据，历史被跳过；group-offsets 从该消费组已提交的位点继续，是持续运行的常规选择。选错会直接造成重复或漏数。
基本信息
| 连接器 | connector=kafka，需要对应版本的连接器 jar 放进 Flink 的 lib 目录 |
|---|---|
| 启动位点 | earliest-offset / latest-offset / group-offsets / timestamp / specific-offsets |
| 数据格式 | json、debezium-json、avro 等，需要相应的格式依赖 |
| 消费组 | properties.group.id，决定位点提交到哪个组 |
| 写入保证 | 结果表可配置交付保证，exactly-once 依赖 checkpoint |
关键配置读法
| 配置项 | 作用 | 注意 |
|---|---|---|
| connector | 连接器类型 | 必须是 kafka，且 jar 版本与 Flink 大版本匹配 |
| topic | 主题名 | 源表与结果表的主题都要显式指定 |
| properties.bootstrap.servers | Broker 地址 | 写两三个即可 |
| scan.startup.mode | 首次读取位置 | group-offsets 依赖已提交位点，首次启动前先确认 |
| format | 数据格式 | 时间字段格式与声明的类型必须匹配，否则整行解析失败 |
典型使用场景
- 实时清洗：读原始主题做字段过滤与规范化，写回处理后的主题
- 流式聚合：按窗口统计指标并写入下游主题供其它系统消费
- 多源合并：把多个主题的数据统一成一种格式后输出
- 实时入仓：把 Kafka 中的数据解析后写入数仓或湖表
- 回放验证：临时用 earliest-offset 重放一段数据验证加工逻辑
Kafka 源表定义
Kafka 源表定义
CREATE TABLE orders_src (
order_id STRING,
amount DECIMAL(10,2),
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND -- 容忍 5 秒乱序
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092',
'properties.group.id' = 'orders_sql',
'scan.startup.mode' = 'group-offsets', -- 从已提交位点继续
'format' = 'json'
);加工并写回 Kafka
加工并写回 Kafka
-- 结果表：写回到另一个主题
CREATE TABLE orders_sink (
order_id STRING, amount DECIMAL(10,2), ts TIMESTAMP(3)
) WITH (
'connector' = 'kafka',
'topic' = 'orders_clean',
'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092',
'format' = 'json',
'sink.delivery-guarantee' = 'exactly-once' -- 需要作业开启 checkpoint
);
-- 持续查询：把清洗后的数据写入结果主题
INSERT INTO orders_sink
SELECT order_id, amount, ts FROM orders_src WHERE amount > 0;连接器 jar 与 Flink 大版本绑定：升级 Flink 时连接器必须同步替换，否则提交时报找不到连接器或方法不存在，这类问题在升级窗口最容易集中出现。
选择 exactly-once 交付保证会影响作业的延迟（需要等待检查点），且必须开启检查点，否则该配置没有意义；对延迟敏感且下游可幂等时，至少一次通常更实用。