Flink读取Kafka数据实战：连接器配置、位点管理与性能调优

简介：一套完整的Flink实时数据处理实战项目打包为zip，覆盖从Kafka消费数据、流式计算到写入Redis集群与MySQL的完整链路。项目面向大数据开发学习者，适合正在搭建实时数仓、需要掌握Kafka Connector、Flink窗口计算以及多种Sink集成的读者。压缩包共145个文件，大小48.47MB，以100个XML配置文件与15个Java源码为主，另有13个class编译产物、4个properties配置、4个lst文件、2个jar依赖以及Maven包装器、gitignore、license等工程文件，目录结构清晰，便于直接导入IDE运行调试。核心代码包含LogEventApp等入口类、日志事件Schema定义、水印提取器及请求响应消息模型，通过具体示例演示了如何配置Kafka消费者组、执行keyBy分组与窗口聚合，并将计算结果通过Redis Sink与JDBC分别写入Redis集群和MySQL数据库，兼顾实时缓存与持久化存储两种典型场景。已有597人学习，对希望快速上手Flink流处理实战、理解实时链路搭建细节的开发者有较高参考价值。

1. 拿到flink读取kafka数据就开跑的人，最后都停在了环境上

有人丢给你一个flink读取kafka数据.zip，解压出来以为改个bootstrap.servers就能跑，结果卡在类冲突、位点报错、数据对不上这三件事上。把 flink 读取 kafka 数据这件事拆到底，真正绕不开的只有三个问题：怎么连上 kafka、怎么把byte[]变成业务对象、怎么让 consumer group 的位点在重启时不回跳也不丢数。这篇不依赖现成的压缩包，从环境、连接器、参数到排错，把这条链路上最常见的坑完整走一遍。适合已经在用 flink sql 或者 DataStream 写实时任务、但还没系统整理过 kafka 消费端细节的人，也适合准备 kafka 面试题时想把 flink 和 kafka 串成体系的人。

2. flink读取kafka数据的前置条件：先有一个能连上的kafka集群

2.1 为什么先确认 kafka 集群，再写 flink 代码

flink 读 kafka，本质上是一个或多个 consumer 实例去 broker 上拉数据。连接串写错、advertised.listeners配错、topic 不存在，这三种情况在 flink 端报的都是超时或无法获取分区，错误信息长得非常像。环境问题不解决，后面做的所有调优都无从谈起。

顺手说一句：kafka 的资料丰富度比 pulsar 高出一截，搜“kafka 安装配置”“kafka 集群安装”，能翻出从 win11 部署到生产集群的各种方案。我一般不建议在 Windows 本地直接跑二进制包，一是文件路径和脚本权限问题多，二是你最终要跑的还是 Linux 和容器，没必要在开发机上多维护一套变量。直接用 docker compose 起一个单节点 KRaft 模式的 kafka，最快，也最接近生产拓扑。这里说的 KRaft 就是 kafka 3.x 之后去掉 zookeeper 的新模式，单进程就能把 controller 和 broker 一起跑起来，部署成本比老的 zookeeper 方案低很多。

2.2 用 docker compose 起 kafka 集群，再创建主题

先给一份可以直接用的docker-compose.yml，单节点 KRaft 模式：

services: kafka: image: bitnami/kafka:3.6 ports: - "9092:9092" environment: - KAFKA_CFG_NODE_ID=0 - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093 - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=1

这段配置里最关键的是KAFKA_CFG_ADVERTISED_LISTENERS。这个地址会写进 broker 返回给客户端的元数据里，flink 客户端拿着这个地址去建立连接。如果 flink 跑在另一个机器上，这里的localhost必须改成 docker 宿主机的内网 IP 或域名，否则你在代码里写host.docker.internal也没用。

启动之后创建 topic：

docker exec -it kafka-kafka-1 kafka-topics.sh \ --bootstrap-server localhost:9092 \ --create \ --topic demo-topic \ --partitions 3 \ --replication-factor 1

--partitions 3这个数字别随便填，它直接决定后面 flink 作业能开多大并行度。topic 的分区数只能增加、不能减少，中途改分区数会让后续所有消费端的分区分配策略重新计算一次，容易引发不必要的 rebalance。开发阶段就按目标并行度来定分区数。

2.3 flink 与 kafka 连接器的版本对齐，这一步不能省

版本不对，flink 读 kafka 数据时最常见的报错是NoSuchMethodError或者ClassNotFoundException，而且这种报错往往出现在env.execute()之后，一旦进入运行期就不好定位了。

我踩过的一个实际坑是：flink 升到 1.18，连接器还在用旧版本，结果WatermarkStrategy传进去不生效，KafkaSource的类倒是能编译，但是运行时消费位点一直重置在 latest，找了一天最后才意识到是连接器版本太旧。所以这一章强调版本对齐，比任何参数调优都优先。

3. 用 DataStream API 让 flink 读取 kafka 数据的最小代码与参数清单

3.1 依赖声明：连接器、格式依赖和冲突排查

先看 pom 里的依赖写法：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>3.2.0-1.18</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-base</artifactId> <version>1.18.1</version> </dependency>

连接器会传递依赖kafka-clients。如果你项目里同时还引了消息中间件客户端、日志采集组件或者别的组件，它们也可能带一份kafka-clients，最终 classpath 里就会出现两个版本。解决办法不是调顺序，而是用mvn dependency:tree把所有传递依赖列出来，看冲突的是哪个版本，然后在 pom 里对冲突依赖用exclusion排除掉。flink 的 kafka 连接器对kafka-clients的版本比较挑，低版本连新版 broker 时api-versions握手会失败。

3.2 最小可运行代码：KafkaSource 的标准姿势

package demo; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.connector.kafka.source.KafkaSource; import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class KafkaReadDemo { public static void main(String[] args) throws Exception { KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("demo-topic") .setGroupId("flink-demo-group") .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 开启 checkpoint，消费位点才能可靠保存 env.enableCheckpointing(60_000); DataStream<String> stream = env.fromSource( source, WatermarkStrategy.noWatermarks(), "kafka-source" ); stream.map(value -> "raw: " + value).print(); env.execute("flink-read-kafka-demo"); } }

这段代码看着简单，但有几个位置值得展开。setStartingOffsets决定的是“第一次启动、kafka 里没有该 group 的提交位点”时的消费起点，不是每次启动都从 earliest 重读。env.enableCheckpointing(60_000)在这里不是备份状态的通用建议，它是 kafka 连接器提交位点的前提：开启 checkpoint 后，flink 才会在 checkpoint 完成时把消费位点作为状态的一部分写到 state backend，同时默认把位点提交回 kafka。如果不开 checkpoint，group.id在 kafka 的__consumer_offsets里就不会有持续更新的提交记录，任务一重启，位点可能回到启动模式指定的位置，表现就是数据重复读或者丢数据。

3.3 KafkaSource 必调的 4 个参数

setStartingOffsets里最容易理解错的是committedOffsets()。它不是直接从 kafka 读最近一次提交的 offset，而是读取“当前消费组在 kafka 里最新的提交”。如果你之前用控制台消费者跑过同一个group.id，那么 flink 的committedOffsets会拿到控制台消费者留下的位点，这会引起数据从某个奇怪的位置开始消费。不同用途的作业，group.id 必须隔离。

3.4 别再用 FlinkKafkaConsumer，换自定义反序列化

很多早期的 flink 菜鸟教程还在用FlinkKafkaConsumer，这个类在 1.15 之后标记为废弃，社区主推的是KafkaSource。两者最大的区别是：FlinkKafkaConsumer把“消费位点管理”和“数据源逻辑”耦合在一起，KafkaSource则把位点初始化、分区发现、反序列化拆成独立模块。生产环境接手老作业遇到FlinkKafkaConsumer时，我一般建议升到KafkaSource，改动量不大。

如果业务字段不止一个字符串，写一个自定义的KafkaRecordDeserializationSchema：

public class OrderSchema implements KafkaRecordDeserializationSchema<Order> { private transient ObjectMapper mapper; @Override public void open(DeserializationSchema.InitializationContext context) { mapper = new ObjectMapper(); } @Override public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<Order> out) throws IOException { Order order = mapper.readValue(record.value(), Order.class); out.collect(order); } @Override public TypeInformation<Order> getProducedType() { return TypeInformation.of(Order.class); } }

注意mapper一定要在open()里初始化，不要在字段声明处直接 new。DataStream 的算子/函数在执行前会经历序列化分发，如果ObjectMapper作为普通成员变量，在分布式环境里可能会被序列化拷到别的 task 上，触发序列化异常。

3.5 并行度与分区数的匹配关系

flink 读 kafka 数据时，一个分区的数据在同一时刻只会被一个 subtask 消费，但一个 subtask 可以消费多个分区。所以并行度大于分区数时，多出来的 subtask 其实是空转，白白占用 slot。并行度小于分区数时，某个 subtask 会处理两个以上分区，一旦其中一个分区流量暴涨，就可能拖慢整个 task。

topic 分区数建议是目标并行度的 1 到 2 倍。生产环境流量波动大时，分区数稍微留点余量，这样 kafka 侧可以做负载均衡，flink 侧也能在资源充足时把并行度提上去。

4. flink sql 读取 kafka 数据：DDL、水位线与 cdc 数据管道

4.1 用 flink sql client 把 kafka topic 当表读

如果不想写 Java，flink 提供 sql client 可以直接把 kafka topic 映射成一张动态表。启动方式很简单：

bin/sql-client.sh

进入 sql client 之后，先建表再查询。这种做法特别适合快速验证数据能否读通、字段映射对不对，不需要走打包部署的完整链路。

4.2 核心 DDL 与 WITH 参数说明

CREATE TABLE kafka_orders ( order_id BIGINT, user_id BIGINT, amount DOUBLE, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'demo-topic', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'flink-sql-group', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' );

scan.startup.mode是 flink sql 里最容易配错的参数，它对应的就是 DataStream API 里的setStartingOffsets：

建表之后，一句聚合查询就能验证数据链路：

SELECT user_id, COUNT(*) AS order_cnt, SUM(amount) AS total_amount FROM kafka_orders GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), user_id;

这个查询里ts是事件时间，窗口是 1 分钟的滚动窗口。kafka 里的数据如果乱序超过 5 秒，WATERMARK会让迟到数据进入下一窗口，因此WATERMARK FOR ts AS ts - INTERVAL '5' SECOND中的 5 秒要根据业务容忍的乱序程度调整。设得太小，晚到数据直接丢；设得太大，窗口计算结果迟迟不发，端到端延迟会明显上升。

4.3 水位线只在需要窗口或 join 时才有意义

flink sql 中 water 相关的坑，主要集中在：数据源本身没有事件时间字段，或者事件时间字段不是TIMESTAMP(3)类型。kafka 消息里的时间字段如果只有毫秒时间戳，DDL 里得先写成BIGINT，再通过TO_TIMESTAMP_LTZ(ts, 3)转成事件时间：

CREATE TABLE kafka_logs ( log_ts BIGINT, log_body STRING, event_ts AS TO_TIMESTAMP_LTZ(log_ts, 3), WATERMARK FOR event_ts AS event_ts - INTERVAL '3' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'log-topic', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'flink-sql-log-group', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' );

之前有人把log_ts直接声明成TIMESTAMP(3)，结果数据都是从 1970 年开始的，窗口聚合结果全是错的。kafka 消息里的时间戳字段如果是BIGINT，必须先显式转换。

4.4 cdc 链路里的 kafka 位置，以及数据血缘怎么查

实时数仓里很常见的一条链路是：flink cdc 采集 MySQL binlog，写入 kafka，再由另一个 flink 作业消费 kafka 做清洗和聚合。这样做的价值是解耦：源库的压力只在 binlog 采集这一层，下游 flink 作业的重启、回溯、扩并发都不会直接打到源库上。

这个链路里 flink 读 kafka 数据的作业，数据血缘可以从作业图里看到大致脉络，但真正精确的字段级血缘需要依赖 flink sql 的 catalog 和 lineage 插件。做数据治理时，我一般建议从 sql 作业的CREATE TABLEDDL 入手，把 kafka topic、字段名、目标表字段一一对应地记录到元数据中心，而不是事后从计算引擎里往回找。

5. flink 读取 kafka 数据变慢或丢数？先查 lag、活性和缓冲

5.1 三个命令、一个页面，定位消费卡点

拿到一个新作业，我做的第一件事永远是看 kafka 侧的消费延迟：

kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group flink-sql-group \ --describe

重点看LAG这一列。LAG 不为 0 不一定代表 flink 出问题了。flink 作业在 checkpoint 对齐阶段会暂停消费，短时间内 LAG 上涨是正常的；但如果 LAG 持续上涨超过 10 分钟，就要去看 flink UI 上 kafka source 的pendingRecords指标。pendingRecords是 flink 已经拉下来但还没处理完的记录数，它和 kafka 的 LAG 之间差距越大，说明数据积压在 flink 内部而不是 kafka 侧。

5.2 消息延迟高和 OOM 的常见原因

OOM 这块有个容易误判的点：kafka 连接器的缓冲是在堆内的，fetch.min.bytes设得太大，会让每次拉取都尝试凑满一个较大的缓冲区，堆积的byte[]在高峰期会占掉不少堆内存。如果业务允许，把fetch.min.bytes从默认的 1 字节调成几百字节量级，能显著减少拉取频率，但对延迟会有一点影响。

5.3 用有限条数验证数据一条不少

调试阶段不要直接拿生产 topic 验证，用kafka-console-producer发有限条数据，再数一遍。

kafka-console-producer.sh \ --bootstrap-server localhost:9092 \ --topic demo-topic

手动发 10 条，flink 端print()出来必须是 10 条。然后回去跑kafka-consumer-groups.sh --describe，LAG应该是 0。这样才能证明从 kafka 到 flink 的链路是闭环的。做对账时，用 sql 作业的COUNT(*)和 kafka 的kafka-run-class.sh kafka.tools.GetOffsetShell两个数字对齐，比肉眼数日志可靠得多。真正稳的消费链路不是看“程序没报错”，而是看消费位点和记录数能不能严丝合缝地咬合。

本文还有配套的精品资源，点击获取