简介一套完整的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_ID0 - KAFKA_CFG_PROCESS_ROLEScontroller,broker - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS0kafka:9093 - KAFKA_CFG_LISTENERSPLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAPCONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - KAFKA_CFG_CONTROLLER_LISTENER_NAMESCONTROLLER - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR1这段配置里最关键的是KAFKA_CFG_ADVERTISED_LISTENERS。这个地址会写进 broker 返回给客户端的元数据里flink 客户端拿着这个地址去建立连接。如果 flink 跑在另一个机器上这里的localhost必须改成 docker 宿主机的内网 IP 或域名否则你在代码里写host.docker.internal也没用。启动之后创建 topicdocker 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.15 及以前内置flink-connector-kafka只能用旧 APIFlinkKafkaConsumer1.16 ~ 1.18独立连接器版本号形如3.2.0-1.18新 APIKafkaSource已经可用写法更干净1.19 及以后连接器单独发版必须去 maven 上看版本矩阵不能随便拿一个版本就塞进依赖我踩过的一个实际坑是flink 升到 1.18连接器还在用旧版本结果WatermarkStrategy传进去不生效KafkaSource的类倒是能编译但是运行时消费位点一直重置在 latest找了一天最后才意识到是连接器版本太旧。所以这一章强调版本对齐比任何参数调优都优先。3. 用 DataStream API 让 flink 读取 kafka 数据的最小代码与参数清单3.1 依赖声明连接器、格式依赖和冲突排查先看 pom 里的依赖写法dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.2.0-1.18/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-base/artifactId version1.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 { KafkaSourceString source KafkaSource.Stringbuilder() .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); DataStreamString 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。如果不开 checkpointgroup.id在 kafka 的__consumer_offsets里就不会有持续更新的提交记录任务一重启位点可能回到启动模式指定的位置表现就是数据重复读或者丢数据。3.3 KafkaSource 必调的 4 个参数参数可选值适用场景setStartingOffsetsearliest()/latest()/committedOffsets()/timestamp(long)首次启动离线回溯用earliest只关心新数据用latestcommittedOffsets适合多作业复用同一个 groupsetCommitOffsetsOnCheckpoints(true)true/false需要外部系统也按 kafka offset 对账时保持默认 truesetProperty(partition.discovery.interval.ms, 60000)毫秒topic 分区数后续扩容时flink 周期发现新分区并自动消费setBounded(STOP_AFTER_CURRENT)或setUnbounded()默认无界流做批量回溯测试STOP_AFTER_CURRENT生产环境用默认无界setStartingOffsets里最容易理解错的是committedOffsets()。它不是直接从 kafka 读最近一次提交的 offset而是读取“当前消费组在 kafka 里最新的提交”。如果你之前用控制台消费者跑过同一个group.id那么 flink 的committedOffsets会拿到控制台消费者留下的位点这会引起数据从某个奇怪的位置开始消费。不同用途的作业group.id 必须隔离。3.4 别再用 FlinkKafkaConsumer换自定义反序列化很多早期的 flink 菜鸟教程还在用FlinkKafkaConsumer这个类在 1.15 之后标记为废弃社区主推的是KafkaSource。两者最大的区别是FlinkKafkaConsumer把“消费位点管理”和“数据源逻辑”耦合在一起KafkaSource则把位点初始化、分区发现、反序列化拆成独立模块。生产环境接手老作业遇到FlinkKafkaConsumer时我一般建议升到KafkaSource改动量不大。如果业务字段不止一个字符串写一个自定义的KafkaRecordDeserializationSchemapublic class OrderSchema implements KafkaRecordDeserializationSchemaOrder { private transient ObjectMapper mapper; Override public void open(DeserializationSchema.InitializationContext context) { mapper new ObjectMapper(); } Override public void deserialize(ConsumerRecordbyte[], byte[] record, CollectorOrder out) throws IOException { Order order mapper.readValue(record.value(), Order.class); out.collect(order); } Override public TypeInformationOrder 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 当表读如果不想写 Javaflink 提供 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配置值行为earliest-offset从每个分区最早的消息开始消费latest-offset从每个分区最新的消息开始消费group-offsets从 kafka 记录的该 group 消费位点开始等价于 DataStream 的committedOffsetstimestamp从指定时间戳之后的消息开始需要额外配置scan.startup.timestamp-millis建表之后一句聚合查询就能验证数据链路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 的常见原因现象先查什么处理方式LAG 涨pendingRecords也涨flink 反压页面背压集中在哪个算子就优化哪个算子不要盲目加并行度LAG 涨pendingRecords很低单条消息过大fetch 缓冲长时间满载调大fetch.max.bytes或检查是否有超大字段频繁 OOM堆内存 / GC 日志kafka source 的缓冲默认占用不算大OOM 多出在下游 keyBy 之后的状态膨胀消费位点频繁回退checkpoint 失败看 checkpoint 超时和失败原因通常是状态太大或后端存储抖动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 --describeLAG应该是 0。这样才能证明从 kafka 到 flink 的链路是闭环的。做对账时用 sql 作业的COUNT(*)和 kafka 的kafka-run-class.sh kafka.tools.GetOffsetShell两个数字对齐比肉眼数日志可靠得多。真正稳的消费链路不是看“程序没报错”而是看消费位点和记录数能不能严丝合缝地咬合。本文还有配套的精品资源点击获取