简介这是一份面向大数据流处理开发者的Flink实战代码包聚焦从Kafka消费实时数据、完成业务计算后分别写入Redis集群与MySQL的完整链路可用于实时监控、日志分析和在线广告等低延迟场景。资源共145个文件约48.47MB以java源码、编译后的class、xml配置为主另含properties配置、jar依赖及mvnw构建脚本便于理解工程结构并直接运行调试。代码演示了事件日志解析、自定义水位线提取、keyBy分组与window窗口统计等关键环节并给出Kafka连接器参数、Redis Sink连接设置以及JDBC批量写入MySQL的实现方式覆盖从数据接入、计算到多端存储的典型流程。适合有一定Flink基础、希望上手真实项目的学习者。目前已有597人学习下载通过研读源码与配置可掌握流式数据落地的常见设计思路和排错要点。1. 拿到一个叫“flink读取kafka数据”的压缩包你真正需要的是什么拿到一个叫 flink读取kafka数据 的压缩包里面大概率是一份可直接运行的 Flink 工程或者是一套配套教程的 demo 源码。这个标题指向的正是实时计算里最常被问到的链路Kafka 里堆着千万级消息Flink 把它们拉下来做处理再交给你下游的存储或告警。它能解决的核心问题不是“能不能读”而是怎么把这条链路从零跑通并且跑得稳。适合刚接手实时任务、想验证 Flink 与 Kafka 连接方式的读者也适合课上跑通了 demo 但没上过生产的同学——代码能起来是一回事参数和坑是另一回事。这篇就按我平时搭这类工程的顺序从环境选型、三种读取方式、调参基线一直说到排障清单。2. 把底座备齐Kafka 与 Flink 的版本选型和本地环境2.1 版本选型JDK、Kafka、Flink 之间怎么搭不翻车排除掉业务逻辑之前先排除版本问题。很多 demo 跑不起来不是代码错了是 Flink 连接器坐标和 Kafka 客户端版本对不上。Flink 1.14 及以前Kafka 连接器叫flink-connector-kafka_2.11或_2.12版本号跟着 Flink 走从 1.15 开始官方把连接器独立发布坐标变成flink-connector-kafka后面的版本号要单独匹配 Flink 主版本。我一般是 Flink 1.17.x 配 Kafka 3.x、JDK 11这套组合踩坑最少。Flink 版本Kafka 连接器坐标写法JDK 建议备注1.14 及以下flink-connector-kafka_2.12:1.14.xJDK 8老接口FlinkKafkaConsumer 还能用1.15 / 1.16flink-connector-kafka:1.15.x/1.16.xJDK 8/11新版 KafkaSource 可用1.17 / 1.18flink-connector-kafka:3.1.0-1.17等JDK 11客户端兼容范围大推荐新工程选这档这里有个容易忽略的点单独引kafka-clients时版本不能低于 Flink 连接器自带的那个否则运行时会报UnsupportedVersionException消息解析全乱。我一般会直接依赖连接器提供的客户端版本不单独引 kafka-clients除非确实需要新版 API。选型别追新Flink 1.18 出了但生态插件跟进有滞后连 Hive、JDBC 这类连接器都以 1.17 为基准更新得更勤。2.2 用 docker-compose 起一个 Kafka 集群的最小配置本地验证不需要真的搭三台机器一个 Kafka 容器足够但要注意新版 Kafka 用了 KRaft 模式不需要单独的 ZooKeeper 容器很多老教程还是双容器方案照抄会白占资源。下面这份 compose 我在多个环境用过拉起来就能被 Flink 访问还带一个 Kafka UI 便于看消息和 lag。services: kafka: image: apache/kafka:3.7.0 container_name: kafka ports: - 9092:9092 environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CONTROLLER_QUORUM_VOTERS: 1localhost:9093 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_NUM_PARTITIONS: 3 kafka-ui: image: provectuslabs/kafka-ui:latest container_name: kafka-ui ports: - 8080:8080 environment: KAFKA_CLUSTERS_0_NAME: local KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092 depends_on: - kafka配置含义说一下KAFKA_NUM_PARTITIONS设为 3是为后面测试 Flink 并行度大于 1 时消息是否打散做准备ADVERTISED_LISTENERS必须写localhost:9092否则容器里启动成功了宿主机上的 Flink 却连不上GROUP_INITIAL_REBALANCE_DELAY_MS设 0 是让消费组尽快开始分配分区测试时不用等那几秒的延迟。跑docker compose up -d后先到 8080 端口看一眼 topic 是否存在再用命令行生产几条数据确认链路通再往下走。2.3 pom.xml 依赖四个必加的坐标与常见冲突拿到 demo 工程第一件事是看 pom.xml不是看代码。以下是我在 Flink 1.17 工程里最常用的最小依赖集能覆盖从读取到执行的完整链路。properties flink.version1.17.2/flink.version maven.compiler.source11/maven.compiler.source maven.compiler.target11/maven.compiler.target /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.1.0-1.17/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-common/artifactId version${flink.version}/version scopeprovided/scope /dependency /dependenciesflink-streaming-java提供 DataStream APIflink-clients负责本地执行环境flink-connector-kafka是读取 Kafka 的核心单独列出来是因为它的版本规则和 Flink 本体不同步。provided作用域要注意在 IDEA 里直接运行 main 方法时provided 的类默认不在运行 classpath 里必须在 Run 配置里勾上 “Include dependencies with provided scope”很多人第一次跑报ClassNotFoundException就是这个原因。3. 读 Kafka 的三种写法从 DataStream API 到 Flink SQL3.1 最简实现KafkaSource 与反序列化器先从一个能直接跑通的最小例子开始。Flink 1.15 之后官方推荐用KafkaSource老接口FlinkKafkaConsumer虽然还在但已经标了 deprecated新工程别再用。这个例子读取一个 topic 的字符串消息打印到控制台。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.environment.StreamExecutionEnvironment; public class KafkaReadDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 开启 checkpointKafka offset 的提交依赖它 env.enableCheckpointing(5000); KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(localhost:9092) .setTopics(demo-topic) .setGroupId(flink-demo-group) .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), kafka-source) .map(value - 收到: value) .print(); env.execute(flink-read-kafka-demo); } }这段代码有三个参数需要解释。setStartingOffsets(OffsetsInitializer.earliest())表示消费组第一次连上 topic 时从头开始读如果你希望只消费新消息改成latest()setGroupId必须显式指定Flink 靠它记录消费位点不写会直接报错setValueOnlyDeserializer(new SimpleStringSchema())只处理消息的 value 部分如果你的消息里 key 也承载业务含义需要换成setDeserializer并实现KafkaRecordDeserializationSchema单纯用 value 是绝大多数 JSON 消息的默认场景。3.2 带 checkpoint 的写法状态与 Exactly-Once 的关系没有 checkpoint 的 KafkaSource 是不可靠的至少对这个标题下的生产场景来说不可用。Flink 的 Kafka 连接器不会在消费到一条消息时就立刻提交 offset而是把提交动作绑定在 checkpoint 完成时——checkpoint 成功offset 才提交。这样设计是为了避免“数据处理了但 offset 没提交重启后重复消费”和“offset 提交了但数据处理失败重启后丢数据”。CheckpointConfig config env.getCheckpointConfig(); config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); config.setMinPauseBetweenCheckpoints(500); config.setCheckpointTimeout(60000); config.setMaxConcurrentCheckpoints(1); config.setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);这段配置里EXACTLY_ONCE是 Flink 状态层面的保证它不等于端到端精确一次——Kafka 到 Flink 这一段靠 checkpoint 能精确但 Flink 写到 MySQL 或 Hive 那段还需要下游配合幂等。setMinPauseBetweenCheckpoints(500)是两次 checkpoint 之间的最小间隔防止频繁快照把磁盘打满RETAIN_ON_CANCELLATION是“后悔药”作业取消后 checkpoint 文件保留下次启动可以从最近一次快照恢复调试血泪经验告诉我这个一定要开。3.3 Flink SQL 方式用 DDL 把 Kafka 当表读如果你的场景是快速验证数据 or 不想写 Java 代码Flink SQL 是更轻的路线。用 Table API 或 sql-client 执行下面这段 DDL就把 Kafka topic 声明成了一张临时表之后可以像查数据库一样查它。CREATE TABLE kafka_demo ( id BIGINT, name STRING, 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-demo, scan.startup.mode earliest-offset, format json );SELECT id, COUNT(*) AS cnt FROM kafka_demo GROUP BY id;SQL 写法的关键参数在 WITH 子句里scan.startup.mode支持earliest-offset、latest-offset、group-offsets语义和 DataStream API 里的setStartingOffsets一一对应format必须是消息真正的序列化格式JSON 和 CSV 最常用如果消息里带 schema 则选debezium-json或canal-json。这里有个坑SQL 方式默认把 Kafka 当无界流GROUP BY会触发状态无限增长生产环境必须配合ttl或窗口使用demo 里无所谓但要知道边界在哪。4. 必调参数与消费稳定性延迟、吞吐、消息顺序三者怎么平衡4.1 从 kafka 消息延迟高说起fetch、linger 与并行度的关系很多人遇到 kafka 消息延迟高第一反应是加 Kafka 节点或调大fetch.min.bytes这往往跑偏。延迟高的根因更多在消费端的 Flink 作业并行度上Kafka 一个分区只会被 Flink source 的一个并行子任务消费如果你 topic 有 12 个分区而 Flink 作业并行度只有 3那理论上再快也只能同时拉 3 个分区的数据消息全堵在 broker 上等分配。先查并行度再看 broker 的fetch.max.wait.ms。参数默认值作用fetch.min.bytes1broker 攒够多少字节才返回调大提升吞吐但抬高延迟fetch.max.wait.ms500等数据的最长时间攒不够字节也会返回max.poll.records500单次 poll 返回的最大记录数max.partition.fetch.bytes1048576单个分区单次拉取上限Flink 连接器的properties里可以直接传 Kafka 参数例如.setProperty(fetch.min.bytes, 1024)。实践上延迟敏感项目把fetch.min.bytes保持默认靠并行度吃吞吐吞吐敏感项目才调大它。有一类延迟是“假延迟”消息量小、broker 一直等数据攒够字节fetch.max.wait.ms到了上限才返回这时调小fetch.max.wait.ms能见效。4.2 消费端多线程与消息顺序性最大化吞吐的代价kafka 消费端多线程如何保证消息顺序性是个高频问题答案很扫兴Flink 里单并行子任务内部严格按 offset 顺序处理一旦并行度大于 1或者你在单并行里再开多线程做异步处理顺序就打散了。顺序不是免费的它和吞吐天然冲突。我一般给业务方的建议就两条一是需要全局有序的场景并行度设 1从源头放弃吞吐二是按 key 局部有序的场景用keyBy(key)让同一个 key 的消息落到同一并行实例这是性价比最高的折中。比如订单变更流只需要同一个订单号的消息有序keyBy(orderId)就能在保住顺序的同时把不同订单分散到多个并行。注意keyBy之后你就不能依赖 Kafka 本身的 offset 顺序了算子内部是按键分组后的顺序。4.3 一套能直接抄的调参基线基于上面这些约束我给出一个通用性较强的初始配置适合日均千万级消息、允许秒级延迟的项目。不是万能公式但是一个能跑的起点。配置项推荐值理由Flink 并行度topic 分区数的 1 到 2 倍保证每个分区都有对应消费线程留冗余应对故障enable.auto.commitfalse避免 Kafka 自动提交和 checkpoint 提交打架checkpoint 间隔30 到 60 秒太频繁磁盘压力大太慢重启恢复丢得多max.poll.records500 到 1000超过处理能力会触发 rebalance反而降吞吐restart-strategyfixed-delay3 次间隔 10 秒防止脏数据让作业无限重启fetch.max.wait.ms500延迟和吞吐的平衡点这套基线跑稳定后再逐步压测调优一次只动一个参数。我见过最典型的翻车就是把所有参数一次性拉满结果是吞吐没上去checkpoint 超时、rebalance 频繁最后把锅甩给 Kafka——这锅 Kafka 不背。5. 避坑打包、序列化、连接器错误与数据积压的排查清单5.1 现象ClassNotFound / NoClassDefFound作业提交后直接失败原因本地 IDEA 跑通了打包传到集群却起不来十有八九是连接器依赖没有打进 jar 包。providedscope 的依赖在集群里有但flink-connector-kafka和kafka-clients这些 runtime 依赖必须跟着作业包走默认 maven package 不会打入。解决用maven-shade-plugin打成 fat jar并把连接器依赖正常打入。plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.5.1/version executions execution phasepackage/phase goalsgoalshade/goal/goals configuration createDependencyReducedPomfalse/createDependencyReducedPom /configuration /execution /executions /plugin检查打包是否成功有个土办法看最终 jar 的大小。只有几十 KB 的肯定没带上连接器正常带 Kafka 连接器的 fat jar 至少几 MB 起步。5.2 现象Flink 的 JDBC 连接器异常——依赖与运行时类加载器冲突原因作业里同时用了flink-connector-jdbc和自己单独引的 MySQL 驱动版本对不上或两个驱动 jar 都进了 classpath类加载器优先加载了旧版本。这类报错花样很多最常见的是ClassNotFoundException: com.mysql.cj.jdbc.Driver或No suitable driver found。解决只保留一个驱动版本且用 shade 插件把驱动 relocate 到一个私有命名空间避免和集群自带的类冲突。relocation patterncom.mysql.cj/pattern shadedPatternshaded.com.mysql.cj/shadedPattern /relocation另一种更干净的思路把 JDBC 连接器和驱动直接放到 Flink 的lib目录作业包里不再打入这样类加载器冲突面最小。按 Flink 官方默认的 child-first 类加载策略放 lib 里的类优先级高于作业包能避免大部分冲突。5.3 现象Sink Hive 表数据不入表作业不报错但查不到数据原因表面是“没写进去”实际常见原因有三个——Hive 方言没开启、分区字段类型或格式不匹配、写入模式不对。我遇到过最隐蔽的是分区字段定义dt STRING而 Flink SQL 里写的是TIMESTAMP两者在 Catalog 里映射成不同类型Hive 建了分区目录但查SELECT *看不到。解决开 Flink SQL 时先执行set table.sql-dialecthive写在INSERT INTO t PARTITION(dt20240601)时显式指定分区值避免让 Flink 推断。还要确认 HiveCatalog 里表 schema 的类型和 Flink 端完全一致别依赖隐式转换。5.4 现象反序列化失败导致作业无限重启offset 卡住不前进原因topic 里混入脏数据比如本该是 JSON 的消息夹了一行纯文本SimpleStringSchema还好说一旦自定义解析抛异常Flink 默认的失败策略是不断重启每次重启从前一个成功 checkpoint 恢复于是那条脏数据被反复消费、反复失败形成一个“永远过不去的坎”。解决配置重启策略限制次数更重要是加侧输出流把解析失败的数据单独接走不做主链路。SingleOutputStreamOperatorString parsed env.fromSource(...) .process(new ProcessFunctionString, String() { Override public void processElement(String value, Context ctx, CollectorString out) { try { JSONObject obj JSON.parseObject(value); out.collect(obj.getString(id)); } catch (Exception e) { // 脏数据走侧输出不打断主流程 ctx.output(new OutputTagString(dirty) {}, value); } } });代码里OutputTag是侧输出流的标签脏数据在这里被分流restart-strategy即使触发了恢复后也能跳过这条消息继续向后消费。这个模式比“无限重启直到人工删 offset”体面得多也一直是生产上我优先推荐的容错姿势。5.5 现象Kafka 消息延迟高——消费端反压与空闲分区原因这一类要区分“源头慢”和“下游慢”。常见的是 sink 写入数据库太慢反压一路传回 sourcesource 不敢继续拉消息lag 自然上涨另一类是 source 并行度大于分区数有一部分子任务分配不到分区lag 分布极度不均。解决先看监控确认是 source 侧还是 sink 侧背压再对症下药——sink 侧慢就优化写入批量、异步、连接池source 侧不均就把并行度调到和分区数一致或更低。用 Kafka UI 或kafka-consumer-groups.sh查 lag 分布一眼能分辨这两种情况所有分区 lag 一起涨是下游反压只有部分分区涨是分配不均。6. 把一条生产级链路串起来读取、转换、写入带状态的目标端前面讲的都是单点能力最后把它们串成一条完整链路Kafka 消费 → 按用户分组统计会话次数 → 写入 MySQL。这个例子几乎覆盖了 flink 读取 kafka 数据 的全部核心动作也是你能从这个 zip 方向里带走的最有价值的东西。public class KafkaToMysqlDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30000); KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(localhost:9092) .setTopics(user-click-topic) .setGroupId(demo-group) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), kafka-source) .map(value - JSON.parseObject(value)) .keyBy(json - json.getString(userId)) .map(new CountAgg()) .addSink(new MysqlSink()); env.execute(kafka-to-mysql-demo); } }keyBy(userId)保证同一个用户的所有点击落到同一并行实例CountAgg里可以用ValueState保存累计值但注意状态会在 checkpoint 里持久化别只把它当普通变量。MysqlSink 里我强烈建议用主键 upsert 而不是 insert这样 Kafka 重复消费时写入 MySQL 是幂等的端到端精确一次才有落点。这个工程方向值得做而且越早把“状态 checkpoint 幂等写入”这套组合想明白后面接 Hive、接 Iceberg、接 CDC 都只是换连接器骨架不会变。我现在接手任何人的老作业第一件事永远是看 checkpoint 配置和 sink 有没有幂等这两点过关作业基本能睡安稳觉。希望帮到你。本文还有配套的精品资源点击获取