简介面向大数据实时处理场景这份资源提供了一套完整的Flink 与 Kafka 集成实战项目适合想要掌握 Flink 流式计算、并打通 Redis 与 MySQL 存储链路的开发者。包内实现了从 Kafka 主题读取数据、进行 keyBy 分组与窗口计算、将结果写入 Redis 集群、再通过 JDBC 批量导入 MySQL 的完整流程同时包含 LogEvent 等典型日志事件处理示例。压缩包共 145 个文件以 Java 源码、编译后 class 文件与 XML 配置为主另有 properties 配置、jar 依赖等辅助内容整体约 48.47MB目录结构便于按模块查阅。已有 597 人学习下载对于需要落地实时计算任务、理解 Flink 连接器与外部存储集成的读者这份资源能提供可运行的项目参照和关键代码思路。1. 这个 zip 包解决的是实时链路的第一公里Flink 读取 Kafka 是实时计算里最常规的一类入口Kafka 里堆着订单、埋点、日志Flink 作业把消息消费进来做清洗、关联、聚合再往下游写。拿到一个叫flink读取kafka数据.zip的工程包本质上就是拿到一套 Flink 加 Kafka 的最小可运行方案里面有源码、pom 或者打好的 jar。这类任务看着简单实际翻车率很高连接器版本没对齐、checkpoint 没开、位点策略选错都会让结果看起来像玄学。下面按解开包先做版本体检、读懂读取层原理、调通参数、绕过生产坑的顺序走一遍。适合正在搭第一个 Flink 实时任务、被版本和参数卡住的工程师。2. 解开 zip 先做版本体检目录、依赖与本地启动解压之后先别急着点 IDE 运行按钮花十分钟做版本体检。这套工程能不能跑起来九成取决于 Flink、连接器、kafka-clients 三者的版本是否咬合。Flink 本体的安装配置到部署不在 zip 范围内这里默认你已经有一个能提交作业的集群或本地环境。2.1 Maven 依赖坐标连接器版本策略的两个时代先看 pom.xml这是整个工程最容易出事的地方。Flink 1.14 及以前Kafka 连接器随 Flink 主版本发版坐标带 scala 后缀dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka_2.12/artifactId version1.14.4/version /dependency这段依赖说明两点_2.12对应 Flink 发行版编译用的 Scala 版本选错会在提交阶段报NoClassDefFoundError: scala/...版本号 1.14.4 必须和你本机装的 Flink 主版本一致差一个小版本都可能在序列化部分出错。Flink 1.15 开始连接器独立发版坐标不再带 scala 后缀dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.1.0-1.17/version /dependency版本号里的-1.17表示配套 Flink 1.17。很多老资料还在贴flink-connector-kafka_2.11这种坐标Flink 1.15 之后已经不再发布对应构件。如果 zip 里还是老坐标你又打算跑在新集群上升级路线就是把连接器换成新版独立坐标并删掉显式声明的 scala 依赖。常见版本对照关系大致如下精确到 patch 版本还是要以mvn dependency:tree为准Flink 版本连接器坐标写法大致内置 kafka-clients1.13flink-connector-kafka_2.12:1.13.62.6 上下1.14flink-connector-kafka_2.12:1.14.42.8.x1.15flink-connector-kafka:3.0.0-1.153.2.x1.16flink-connector-kafka:3.0.1-1.163.2.x1.17flink-connector-kafka:3.1.0-1.173.4.x1.18flink-connector-kafka:3.2.0-1.183.5.x一点提醒连接器自带的 kafka-clients 版本不必和 broker 版本严格一致但别跨大版本太多比如用 2.x 的客户端连 3.x 的 broker虽然多数情况下能跑遇到新协议特性就会出奇怪问题。一份能直接抄的 pom 关键段如下properties flink.version1.17.2/flink.version /properties dependencies !-- 集群运行时已提供本地编译需要 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency !-- 连接器必须打进作业 jar -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.1.0-1.17/version /dependency /dependenciesflink-streaming-java标provided因为 Flink 集群的 lib 目录里已经有这些类打 fat jar 时带上反而容易冲突flink-connector-kafka不标 provided它必须随作业包一起提交。这个取舍是 Flink 工程里最常见的约定zip 里的 pom 如果不是这么标的提交后大概率要踩 5.1 的坑。提示连接器版本高于 Flink 主版本时往往还能兼容但低版本连接器跑高版本 Flink常常在 JobGraph 构造阶段报找不到方法签名。症状在 UI 里不明显日志里会带NoSuchMethodError的混合信息排查时先怀疑版本矩阵。2.2 解包后的目录结构源码包还是遗产 jar标准 Maven 工程解开后结构大致是这样flink-read-kafka-demo/ ├── pom.xml ├── src/main/java/com/demo/kafka/ │ ├── KafkaReadingJob.java │ └── KafkaConsumerJobRunner.java ├── src/main/resources/ │ └── log4j2.properties └── README.md如果 zip 里只有一个 jar 没有源码用jar tf看包内结构jar tf flink-read-kafka-demo.jar | grep -E kafka|KafkaReadingJob | head -30重点看两处org/apache/flink/connector/kafka路径存在与否决定连接器有没有被打进去META-INF/MANIFEST.MF里有没有Main-Class。缺前者说明是瘦包提交到集群会因为缺类直接失败没有Main-Class也能跑提交时用-c参数指定主类即可。另外检查src/main/resources里有没有log4j2.properties或logback.xml。Flink 默认用 log4j2作业里如果带了 logback 的 class 但没带对应实现日志会静默消失排查时看不到任何输出容易被误判成「没消费到数据」。这种问题不算少见尤其在整合 Spring Boot 依赖时最容易碰上。2.3 本地跑通的最小命令打包、提交、看日志本地验证的前提是有一个能连上的 Kafka。没有独立集群时起一个单机容器是最快的docker run -d --name kafka -p 9092:9092 \ -e KAFKA_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092 \ apache/kafka:3.7.0这个镜像启动后自带 KRaft 模式无需单独起 ZooKeeper。KAFKA_ADVERTISED_LISTENERS必须配成客户端能访问的地址这里容器端口映射到宿主机 9092所以写localhost:9092即可如果作业跑在另一个容器里这个值就得改成宿主机 IP 或服务名细节见第 5.4 节。打包和提交命令是mvn clean package -DskipTests flink run -d \ -c com.demo.kafka.KafkaReadingJob \ -p 2 \ target/flink-read-kafka-demo.jar-c指定入口类-p 2设置并行度。提交后立刻看日志tail -f log/flink-*-taskexecutor-*.log | grep -i order-topic\|Exception如果作业在 YARN 或者 K8s 上日志路径不一样但「确认 TaskManager 起来、Source 线程跑起来」这个思路不变。看到 Web UI 的 Job 拓扑里出现Kafka Source节点才算真正跑通。这里有个细节-p指定的并行度决定 Source 算子起多少个并行的 Kafka 消费者线程一般不要超过 topic 总分区数否则会有消费者线程空转。3. 读取层要选对FlinkKafkaConsumer 与 KafkaSource 的取舍讲两个 API 之前先回答一个必然会被问的问题Kafka 自带的消费者客户端也能写 while 循环实现同样的效果为什么还要 Flink 来读因为手写循环意味着要自己处理分区分配、offset 持久化、故障重启、并行度扩容这些是分布式消费组的完整协议Kafka 客户端暴露的是单机视角。Flink 连接器把消费组协议接进 Source 框架业务代码不用再管「消费到哪了、断了怎么办」。Kafka 原理上是一个分布式提交日志Flink 读数据本质上就是消费这个日志区别只在谁来替你管位点。3.1 老连接器 FlinkKafkaConsumeroffset 与 checkpoint 的绑定关系FlinkKafkaConsumer 是 KafkaSource 出来之前的正统写法大量老工程还在用它。它最核心的机制是作业开启 checkpoint 后每个分区的消费位点作为算子状态的一部分随 checkpoint 快照一起持久化。作业失败时从最近一次成功的 checkpoint 恢复这样才能做到 at-least-once配合下游幂等写入还能达到 exactly-once。这个机制有个容易被忽略的前提checkpoint 不开启这一切都不成立。不开启时 Flink 不会主动提交 offsetKafka 消费者会按enable.auto.commit的默认值 true 周期提交提交间隔 5 秒作业一挂就丢位点。老 API 的写法是addSourceProperties props new Properties(); props.setProperty(bootstrap.servers, localhost:9092); props.setProperty(group.id, flink-demo-group); FlinkKafkaConsumerString consumer new FlinkKafkaConsumer( order-topic, new SimpleStringSchema(), props ); consumer.setStartFromLatest(); consumer.setCommitOffsetsOnCheckpoints(true); DataStreamString stream env.addSource(consumer);setCommitOffsetsOnCheckpoints(true)决定 checkpoint 完成时Flink 是否把位点提交回 Kafka broker。这个开关不影响 Flink 内部的故障恢复但影响「同一个 group.id 下次被别的作业消费时的起点」。很多人只开 checkpoint 不设这个参数结果换程序复用同一个 group.id 拉历史发现位置完全不对查半天不是代码问题而是位点提交方向的问题。3.2 新连接器 KafkaSource为什么 1.15 之后官方都推荐它Flink 1.14 引入 KafkaSource1.15 起官方文档把 FlinkKafkaConsumer 标为 deprecated新代码建议一律用 KafkaSource。表面原因是接口统一到 FLIP-27 Source API实际收益有三个。第一个是「有界/无界」显式化。FlinkKafkaConsumer 只能无界消费KafkaSource 通过setBounded能消费到指定位置就结束这让补数据和批流一体变得可行。第二个是动态分区发现变成一等公民老 API 要在 Properties 里传flink.partition-discovery.interval-millis字符串属性写错一个字母就静默失效新 API 用 builder 方法类型安全写错会编译报错。第三个是反序列化器的表达更清晰setValueOnlyDeserializer和setDeserializer一眼能看出是只解 value 还是 key/value 都解。值得提醒的是经常有人把 flink-connector-kafka 和 Flink CDC 那条链路混在一起。Kafka 连接器消费的是已经落进 Kafka 的消息Flink CDC 是直连数据库扫 binlog 的依赖、部署、权限模型完全不是一回事。如果你的 zip 里出现了 cdc 相关依赖那说明它可能还附带另一个需求先拆清楚再动代码别指望一个作业同时干两件事。3.3 消费位点从哪开始earliest、latest、committedOffsets 怎么选位点初始化只在一个条件下生效该 group.id 在 Kafka 里没有已提交 offset 时。之后的重启恢复都走 checkpoint 状态不再走这三个策略。这个前提理解了就不会在「为什么改了 earliest 没效果」上耗时间。位点策略辨析也是 Flink 面试题里最常见的一问。OffsetsInitializer.earliest()从分区最早可用 offset 开始适合首次上线要补全量历史数据的场景。注意是「最早可用」Kafka 按 retention 清掉过期消息后最早的可能是 3 天前而不是 0。OffsetsInitializer.latest()上线瞬间之后的新消息才要历史数据一条不碰。适合只关心实时增量的业务。committedOffsets()从该 group 上次提交的位置继续是最稳妥的恢复策略。Kafka 里没有历史提交时需要兜底参数常见写法是committedOffsets(OffsetsInitializer.latest())避免程序第一次跑就把历史全量拉一遍。OffsetsInitializer.timestamp(long)从某个时间点之后的消息开始消费。这是数据补拉、故障回溯的后悔药比如昨晚数据算错今天想用同一套代码重新消费昨天的数据这个策略最合适。我的习惯是生产作业统一用committedOffsets(OffsetsInitializer.latest())兜底首次上线如果需要补历史显式改一次 earliest 并写注释跑完再改回来。这样留痕后面接手的人不用猜当初为什么这么设。Kafka 里如果历史偏移已经过期committedOffsets 会按兜底走这正好解决「消费者 group 过期被删除、再上线时冷启动」的场景。4. 核心实现一个能上生产的 KafkaSource 作业这一章把 zip 里最值钱的部分讲透。很多 Flink 实时计算 Demo 都用词频统计做初体验换成业务消息也一样核心就是 source 到算子再到 sink 的链路。这里的模板直接替换业务逻辑就能用。4.1 主程序从 KafkaSource 到 Sink 的最小闭环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; import java.time.Duration; public class KafkaReadingJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // checkpoint 是 offset 管理的基石这里设 10 秒一次 env.enableCheckpointing(10_000); KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(localhost:9092) .setTopics(order-topic) .setGroupId(flink-order-group) // 生产建议 committedOffsets(OffsetsInitializer.latest()) 兜底 .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .setPartitionDiscoveryInterval(Duration.ofMinutes(5)) .build(); DataStreamString stream env.fromSource( source, WatermarkStrategy.noWatermarks(), Kafka Source ); stream.map(value - { // 这里替换成实际业务JSON 解析、字段清洗、关联维表 System.out.println(value); return value; }).name(stdout-log); env.execute(flink-read-kafka-demo); } }逻辑说明env.fromSource接收三个参数——source、水位线策略、算子名字。水位线策略这里先用noWatermarks()占位做窗口聚合必须换成真正的策略见第 6.2 节。map里先打点确认数据到达再接实际业务。env.execute才是真正把作业图提交给集群之前所有算子只是构建了一个 DAG不调用它作业根本不会跑。参数说明setPartitionDiscoveryInterval(Duration.ofMinutes(5))让作业每隔 5 分钟检查一次 topic 是否有新分区如果 Kafka 侧在运行中给 topic 添加分区不设这个参数新分区永远不会被消费到。setStartingOffsets(OffsetsInitializer.earliest())这里写死 earliest 是为了本地测试能看到完整消息流生产环境改成 committedOffsets 兜底 latest理由在第 3.3 节。4.2 必调参数把 zip 里的默认配置过一遍拿到 zip 之后第一件事是逐个检查 Properties 或 builder 里的参数不要默认原作者写的就是对的。下面是这类作业在生产环境最常改的 7 个参数参数默认值生产建议说明bootstrap.serverslocalhost:9092broker 地址逗号分隔至少写 2 个只写 1 个 broker它下线后无法更新元数据group.id无业务名.topic.用途决定消费位点归属上线后不要随便改enable.auto.committruefalse由 Flink checkpoint 统一提交避免重复消费auto.offset.resetlatestearliest 或 latest按首次上线需求定只在无已提交位点时生效flink.partition-discovery.interval-millis不启用3000005 分钟老 API 传 Properties新 API 用 builder 方法fetch.min.bytes11024-65536调大减少请求次数但会增加单批等待fetch.max.wait.ms500500-2000低流量 topic 的延迟天花板与上一条联动enable.auto.commitfalse这条要重点解释。Flink 连接器内部虽然会忽略一部分客户端自动提交设置但如果你用的是新 API 且没开 checkpointFlink 不会替你提交 offset这时候把 auto.commit 关掉重启时 offset 会退回 Kafka 保留的最近一次提交点产生重复消费。反过来开了 checkpoint 又开着 auto.commit可能出现 Flink 恢复和客户端提交竞争位点忽前忽后。另一个隐藏参数是max.poll.records它默认 500。它不写在 Flink 连接器的 builder 里而是 kafka-clients 的配置通过 Properties 设置。这条参数调太大会在慢消费下触发 5.5 的问题。4.3 打包与提交Fat Jar 的三种事故现场提交作业的标准姿势是打 fat jar用 shade 插件把连接器、反序列化器、业务依赖全部并进去排掉集群已经有的 Flink 运行时。2.1 节的 pom 里已经给了 filters这里补三件打包时必然遇到的事。第一件shade 出来的 jar 里有签名文件不排除会在提交时报SecurityException: Invalid signature file digest。filter 里排除META-INF/*.SF、META-INF/*.DSA、META-INF/*.RSA是标准动作。第二件如果用了 Scala 2.12 的 Flink又往 fat jar 里塞了 flink-streaming-scala会和集群自带的 scala-library 冲突表现为奇怪的NoSuchMethodError。第三件打成 uber-jar 后可以先验证再提交java -jar target/flink-read-kafka-demo.jar --help如果入口类没配 Main-Class这条命令会直接报no main manifest attribute说明打包配置里少了Main-Class比提交到集群再失败省时间。验证通过后提交提交命令里也能覆盖并行度flink run -d -c com.demo.kafka.KafkaReadingJob \ -p 4 target/flink-read-kafka-demo.jar提交后到 Web UI 的 Job 页面看两个指标Records Received和Records Sent。这两个值持续增长说明 source 和 sink 都活着。如果Records Received一直是 0先确认 topic 里有数据别急着怀疑并行度。5. 生产环境最容易踩的 5 个坑现象、原因与修复这几个坑是几轮线上事故换来的血泪经验按出现频率从高到低排。每一条都按「现象 → 原因 → 解决」的路径讲方便你直接对照。5.1 NoClassDefFoundError依赖被覆盖是头号杀手现象本地 IDE 跑得好好的提交到集群立刻失败日志里出现NoClassDefFoundError: org/apache/kafka/clients/consumer/ConsumerConfig或者NoSuchMethodError指向某个 Flink 算子方法。原因fat jar 里 kafka-clients 版本和集群 lib 目录里的版本不一致类加载时后加载的覆盖先加载的或者 flink-streaming-java 没有标 provided被一起打进 jar和集群自带的同包名类互相打架。解决先看依赖树mvn dependency:tree -Dincludesorg.apache.kafka:kafka-clients确认唯一的版本再把 flink-streaming-java、flink-runtime 这类坐标全部标为 provided最后用 shade 排除签名文件。三步做完绝大多数类冲突消失。这条经验适用于 zip 里所有自带依赖的工程不只是 Kafka 连接器。5.2 脏数据打崩作业反序列化要做容错层现象作业运行几分钟后开始持续重启日志里是SerializationException: Error deserializing key/value for partition或者反序列化成功但处理函数里 NPE。原因Kafka topic 里混入了非 UTF-8 的字节、空消息或者 JSON 结构变化。SimpleStringSchema对 null value 的处理是直接上抛业务字段缺失后代码里又没有空值保护。解决在 source 之后立刻接一层「清洗 map」包住解析逻辑异常消息计数后丢弃或写入死信 topic而不是让整个作业失败stream.map(value - { try { return parseJson(value); } catch (Exception e) { // 记录异常类型和消息前缀避免把整条大消息打爆日志 System.err.println(e.getClass().getSimpleName() : truncate(value, 200)); return null; } }).filter(Objects::nonNull);注意map里抛异常默认会让作业失败所以必须 catch返回 null 再由 filter 过滤掉。这套「先保命再治理」的思路比让作业直接挂掉再人工捞日志排错划算得多。生产上建议把异常计数接入 metric报警比日志有价值。5.3 重复消费checkpoint 不开全白干现象作业每次重启后都会重新消费最近几分钟的数据下游出现重复记录代码逻辑又没有变化。原因十有八九是 checkpoint 没开。Flink 连接器只有在 checkpoint 开启时才把 offset 维护进状态后端没开时 offset 的位置由 Kafka 客户端自动提交提交间隔内崩溃就回到旧位点。还有一种情况是 checkpoint 开了但一直失败重启策略又没配作业从初始位点重新拉。解决env.enableCheckpointing(10_000)是底线同时配重启策略env.enableCheckpointing(10_000L, CheckpointingMode.EXACTLY_ONCE); env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10_000L));fixedDelayRestart(3, 10s)表示最多试 3 次每次间隔 10 秒配完后 checkpoint 失败不会再直接从位点 0 拉全量。注意 checkpoint 的 state 后端也要配默认的 MemoryStateBackend 在作业挂掉后状态就丢了等于没开。生产环境至少用文件系统后端或者 RocksDB。5.4 容器里连不上 Kafkalocalhost 地址玄学现象代码在本地机器上一切正常打到 Docker 或 K8s 里就报connect timed out、Connection refused或者一直卡在Initializing connection。原因Kafka 的advertised.listeners对外发布的是 broker 自身的视角。如果发布的是PLAINTEXT://localhost:9092容器里的作业访问 localhost 指向自己当然连不上。这不是 Flink 的问题是 Kafka 网络模型的问题。解决先分清两个地址——listeners是 broker 监听的地址advertised.listeners是广播给客户端的地址。作业能访问哪个advertised 就写哪个。单机容器映射时写宿主机 IPK8s 里写 Service 的 DNS 名。改完 Kafka 配置后用nc -vz broker-ip 9092从作业所在的网络环境验证一次通得过再怀疑 Flink 代码。这段「玄学」其实全是网络视角不一致。5.5 被踢出消费组max.poll.records 不是越大越好现象作业没有报错但消费速度越来越低日志里出现Member ... failed to send heartbeat随后 group 开始 rebalance消费位点倒退重复消费一批消息。原因Kafka 消费者规定两次 poll 的间隔不能超过max.poll.interval.ms默认 300 秒。处理线程在map里耗时太长或者单批拉取量太大导致下一次 poll 迟迟不来broker 判定消费者死亡并踢出 group。把max.poll.records调大只是为了追求吞吐实际上放大了这个风险。解决调小max.poll.records到 100-200同时把max.poll.interval.ms适当放宽到 600 秒但要同时监控单条消息的处理时延。另一个隐蔽原因是 Sink 端写数据库慢背压直接传导到 sourcepoll 间隔被拖长。个别 topic 分区特别多而并行度小时单个 task 要么饿死要么超载优先用kafka-consumer-groups.sh --describe看每个分区的 lag再决定加并行度还是拆 topic。6. 从跑通到可维护验证三步与一个水位线技巧6.1 验证三步先看指标再信日志作业提交后用三段式验证确认整条链路真的通。第一步从 Kafka 生产端发送两条带特征值的测试消息echo {orderId:TEST-001,amount:99} | \ kafka-console-producer.sh --bootstrap-server localhost:9092 \ --topic order-topic第二步等 10 秒打开 Flink Web UI看 source 算子的Records Received是否 2。这里不要只看 TaskManager 日志日志经常因为 logback 冲突或输出等级设置而丢消息。第三步去 sink 的目标位置确认TEST-001出现。如果是 Kafka 写 Kafka用kafka-console-consumer.sh反向确认如果想看得更直观也可以接一个 Kafka 可视化工具看 topic 里的消息和 lag比命令行容易懂。6.2 一个必改的进阶参数事件时间水位线很多从 zip 直接抄跑的作业窗口任务永远不触发根因都是 Watermark 策略被写成noWatermarks()。如果消息里有业务时间戳比如订单创建时间ts正确写法是stream.assignTimestampsAndWatermarks( WatermarkStrategy.StringforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((value, ts) - parseEventTime(value)) );forBoundedOutOfOrderness(5s)表示容忍事件时间最多乱序 5 秒窗口在「事件时间超过窗口结束时间 5 秒」后才触发。不写这个策略用noWatermarks()时处理时间窗口不受影响但事件时间窗口永远不会触发而且不会报错。参数定好之后建议把 zip 里的 Properties 和 builder 参数全部收拢到一份配置里版本、位点策略、并行度、checkpoint 间隔作为部署参数管理而不是散落在代码里。我见过太多作业因为 group.id 散写在不同 main 方法里换人维护后改了一处漏了另一处直接导致线上重复消费。从那以后我的习惯是Kafka 消费作业参数必须集中配置位点策略和 group.id 每次变更都写 changelog。希望帮到你。本文还有配套的精品资源点击获取