Flink SQL-Client Kafka connector

下载对应版本的 Kafka 毗连器 JAR 文件（如 flink-sql-connector-kafka-.jar），并放置到 Flink 的 lib/ 目次下。
Source

1. CREATE TABLE kafka_source (
2.     user_id STRING,
3.     event_time TIMESTAMP(3),
4.     action STRING,
5.     -- 定义事件时间与 Watermark
6.     WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
7. ) WITH (
8.     'connector' = 'kafka',                      -- 指定连接器类型
9.     'topic' = 'test',                    -- Kafka Topic 名称
10.     'properties.bootstrap.servers' = 'chb1:9092',  -- Kafka Broker 地址
11.     'properties.group.id' = 'flink-sql-group',  -- 消费者组 ID
12.     'scan.startup.mode' = 'earliest-offset',    -- 初始消费位点（可选：latest-offset）
13.     'format' = 'json'                           -- 数据格式（JSON、Avro 等）
14. );

复制代码

Sink

2. drop table kafka_sink;
3. CREATE TABLE kafka_sink (
4.     user_id STRING,
5.     window_start TIMESTAMP(3),
6.     action_count BIGINT
7. ) WITH (
8.     'connector' = 'kafka',                      -- 连接器类型
9.     'topic' = 'output_topic',                   -- 目标 Topic
10.         'properties.group.id' = 'ksink-01',
11.     'properties.bootstrap.servers' = 'chb1:9092',
12.     'properties.auto.offset.reset' = 'earliest',
14.     'format' = 'json',                          -- 数据格式
15.     'sink.partitioner' = 'round-robin',         -- 分区策略（round-robin、key-hash）
16.         -- 事务
17.         'properties.transaction.timeout.ms'='300000',
18.     'sink.transactional-id-prefix' = 'ksink-tid-',       
19.     'sink.delivery-guarantee' = 'exactly-once'  -- 投递语义（需开启 Checkpoint）
20.        
21. );

复制代码

写入到Sink

1. -- 按窗口统计用户行为次数
2. INSERT INTO kafka_sink
3. SELECT
4.     user_id,
5.     TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
6.     COUNT(action) AS action_count
7. FROM kafka_source
8. GROUP BY
9.     user_id,
10.     TUMBLE(event_time, INTERVAL '1' MINUTE);

复制代码

题目

1、 The transaction timeout is larger than the maximum value allowed by the broker (as configured by transaction.max.timeout.ms).
在 Flink SQL 中设置 Kafka 的 transaction.timeout.ms 时，需通过 properties. 前缀转达 Kafka Producer 的专属参数。
sink 端添加'properties.transaction.timeout.ms'='300000',
2、查询kafka_sink报错： Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partitions: [output_topic-0]
Flink SQL大概未精确设置auto.offset.reset参数，导致无法找到偏移量。您须要在Flink SQL的WITH子句中明确指定auto.offset.reset的值

1. 'properties.auto.offset.reset' = 'earliest',

复制代码