Data Engineering Zoomcamp Kafka 理论篇从流处理基础到 Kafka Streams 实战Java 示例全解析【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp本指南是 Data Engineering Zoomcamp 课程「流处理」模块中的可选理论章节07-streaming/theory/README.md系统讲解 Kafka 的核心概念、Producer/Consumer 机制与配置要点并深入到 Kafka Streams 的流式聚合、流表 Join、时间窗口与 Schema Registry 等进阶主题。文章以仓库内 kafka_examples 的完整 Java 工程为主线逐个拆解每个示例类的设计意图、关键配置与可运行方式读完你不仅能理解 Kafka 理论还能直接上手运行这套 producer/consumer/KStream 示例并掌握用 TopologyTestDriver 做流处理单元测试的方法。一、章节定位这门课程里 Kafka 扮演什么角色在 07-streaming 模块中课程主线使用的是 Redpanda 与 Python见 07-streaming/README.md 及 07-streaming/02-redpanda.md 等章节。而 Kafka theory (optional) 这一小节属于可选的补强内容它以视频讲座 Java 代码示例的形式讲解 Kafka 本身的概念体系帮助学有余力的读者理解流处理的理论底座。整个小节分为两大部分部分覆盖主题对应章节编号Stream processing流处理概念、Kafka 是什么、Confluent Cloud、Producer/Consumer、Kafka 配置7.0.1 ~ 7.6Kafka Streams流处理基础、流 Join、流测试、时间窗口、ksqlDB/Connect、Schema Registry7.7 ~ 7.12配套资源包括Stream processing 幻灯片、Kafka Streams 幻灯片、Kafka Configuration Reference 与 Streams Concepts。代码示例集中在 07-streaming/theory/java/kafka_examples 这一 Gradle 工程中。二、示例工程总览一个完整的 Kafka Java 教学项目07-streaming/theory/java/kafka_examples 是一个基于 Gradle 的独立 Java 工程目录结构如下07-streaming/theory/java/kafka_examples/ ├── settings.gradle # 工程名 kafka_examples ├── gradlew / gradlew.bat / gradle/ # Gradle Wrapper跨平台免安装运行 └── src/ ├── main/ │ ├── avro/ # Schema Registry 用 Avro Schema │ │ ├── rides.avsc # 基础 RideRecord Schema │ │ ├── rides_compatible.avsc # 兼容演进示例新增可空字段 │ │ └── rides_non_compatible.avsc# 非兼容演进示例 │ └── java/org/example/ │ ├── Topics.java # 主题名常量定义 │ ├── Secrets.java # 集群与 Schema Registry 凭据占位 │ ├── JsonProducer.java # 读取 CSV 并发布 JSON 消息 │ ├── JsonConsumer.java # 订阅并消费 JSON 消息 │ ├── AvroProducer.java # 使用 Avro Schema Registry 发布 │ ├── JsonKStream.java # KStream 分组计数 │ ├── JsonKStreamWindow.java # KStream 时间窗口计数 │ ├── JsonKStreamJoins.java # KStream 流-流 Join │ ├── customserdes/CustomSerdes.java # 自定义 Serde 工厂 │ └── data/ # Ride、PickupLocation、VendorInfo └── test/java/org/example/ ├── JsonKStreamTest.java # 计数拓扑单元测试 ├── JsonKStreamJoinsTest.java # Join 拓扑单元测试 └── helper/DataGeneratorHelper.java工程配置在 settings.gradle 中只有仓库声明与工程名定义具体依赖通过 Gradle Wrappergradlew解析因此任何安装了 JDK 的机器都能以./gradlew一键构建运行无需预先安装 Gradle。工程把所有主题名收敛在 Topics.javapublic class Topics { public static final String INPUT_RIDE_TOPIC rides; public static final String INPUT_RIDE_LOCATION_TOPIC rides_location; public static final String OUTPUT_TOPIC vendor_info; }而集群凭据集中在 Secrets.java 中以占位符形式给出运行前必须替换为真实值public class Secrets { public static final String KAFKA_CLUSTER_KEY REPLACE_WITH_YOUR_KAFKA_CLUSTER_KEY; public static final String KAFKA_CLUSTER_SECRET REPLACE_WITH_YOUR_KAFKA_CLUSTER_SECRET; public static final String SCHEMA_REGISTRY_KEY REPLACE_WITH_SCHEMA_REGISTRY_KEY; public static final String SCHEMA_REGISTRY_SECRET REPLACE_WITH_SCHEMA_REGISTRY_SECRET; }从源码结构可以推断教学数据的载体是资源目录下的rides.csv被 JsonProducer.java 与 AvroProducer.java 通过getResource(/rides.csv)加载数据字段与纽约出租车记录对应例如 VendorID、乘客数、行程距离等。三、Stream processing 核心概念Kafka 到底是什么3.1 流处理的两步认知按视频章节 7.0.17.0.2 的脉络先建立两个基本概念流Stream无界、持续到达的事件序列事件之间带有时间顺序与因果联系流处理Stream Processing在事件飞行途中持续进行计算过滤、聚合、关联、窗口统计等区别于批量处理攒一批再算的模式。本仓库主线模块07-streaming采用 Redpanda 作为 Kafka API 兼容的实现来落地这些概念见 07-streaming/02-redpanda.md而理论篇则回到 Apache Kafka 原生视角用 Java 展示同一套机制的底层形态。3.2 Kafka 的基本模型视频 7.3 介绍的 Kafka 核心组件包括Topic主题消息的逻辑分类生产者写入、消费者订阅的单元Partition分区每个 Topic 被切成多个分区以水平扩展与并行消息在分区内有序Producer生产者向指定 Topic 发布消息Consumer消费者从 Topic 拉取消息多个消费者组成 Consumer Group 分摊分区Broker代理节点Kafka 集群中负责存储与转发消息的服务器。配套的 07-streaming/theory/java/kafka_examples 工程恰好用rides打车行程、rides_location上车地点、vendor_infoJoin 输出三个主题把这些模型一一落到了可运行的代码里。3.3 用 Confluent Cloud 还是自建集群视频 7.4 讲解了 Confluent Cloud——云托管的 Kafka 服务。示例代码中大量出现pkc-75m1o.europe-west3.gcp.confluent.cloud:9092这样的集群 bootstrap 地址见 JsonProducer.java 等说明课程示例默认跑在 Confluent Cloud 上。你可以注册试用账号创建集群然后在 Secrets.java 中填入 API Key / Secret并把 bootstrap 地址替换为你自己集群的地址。需要说明的是这些集群地址、端点与密钥均为课程当时的示例占位值实际运行时必须按自己创建的集群修改。四、Producer 与 Consumer 实战JSON 消息的发送与接收4.1 JsonProducer把 CSV 变成 Kafka 消息JsonProducer.java 完整演示了一个生产者的生命周期其核心配置如下props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, pkc-75m1o.europe-west3.gcp.confluent.cloud:9092); props.put(security.protocol, SASL_SSL); props.put(sasl.jaas.config, org.apache.kafka.common.security.plain.PlainLoginModule required username Secrets.KAFKA_CLUSTER_KEY password Secrets.KAFKA_CLUSTER_SECRET ;); props.put(sasl.mechanism, PLAIN); props.put(client.dns.lookup, use_all_dns_ips); props.put(session.timeout.ms, 45000); props.put(ProducerConfig.ACKS_CONFIG, all); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaJsonSerializer);逐项解读这批配置对应视频 7.6 Kafka configuration 的主题配置项值作用bootstrap.servers集群地址:9092建立初始连接用的 broker 地址列表security.protocolSASL_SSL使用 SSL 加密传输 SASL 认证sasl.mechanismPLAIN采用用户名/密码式的 PLAIN 认证机制sasl.jaas.configPlainLoginModule ...提供认证所需的用户名与密码来自 Secretsclient.dns.lookupuse_all_dns_ips当 DNS 返回多个 IP 时全部用于连接尝试提升可用性session.timeout.ms45000会话超时时间超过则被视为故障acksall等待所有副本确认后才认为写入成功换取最强持久性保证key.serializer/value.serializerStringSerializer/KafkaJsonSerializer键用字符串序列化值用 JSON 序列化数据读取使用 OpenCSV 的CSVReader跳过表头后逐行把字符串数组映射为Ride对象。发布时做了两处有教学意义的设计时间戳改写ride.tpep_pickup_datetime LocalDateTime.now().minusMinutes(20)把静态 CSV 数据的时间字段改写成相对现在的时间保证下游时间窗口与 Join 逻辑都依赖时间差有真实的近期数据可算Key 选择以String.valueOf(ride.DOLocationID)下车地点 ID作为消息 Key并用record.get().offset()打印返回的偏移量同时以 500ms 间隔Thread.sleep控制发送节奏便于观察。4.2 JsonConsumer订阅并拉取消息JsonConsumer.java 演示消费者的标准配置props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, pkc-75m1o.europe-west3.gcp.confluent.cloud:9092); props.put(security.protocol, SASL_SSL); props.put(sasl.jaas.config, org.apache.kafka.common.security.plain.PlainLoginModule required username Secrets.KAFKA_CLUSTER_KEY password Secrets.KAFKA_CLUSTER_SECRET ;); props.put(sasl.mechanism, PLAIN); props.put(client.dns.lookup, use_all_dns_ips); props.put(session.timeout.ms, 45000); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaJsonDeserializer); props.put(ConsumerConfig.GROUP_ID_CONFIG, kafka_tutorial_example.jsonconsumer.v2); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); props.put(KafkaJsonDeserializerConfig.JSON_VALUE_TYPE, Ride.class);与生产者对称消费者关注的是反序列化与消费语义key.deserializer/value.deserializer键反序列化为 String值反序列化为 JSON 对象group.id消费者组 ID同一组内的消费者会分摊分区auto.offset.reset earliest当组没有已提交偏移量时从最早的消息开始读KafkaJsonDeserializerConfig.JSON_VALUE_TYPE Ride.class告诉 JSON 反序列化器目标 POJO 类型把 JSON 直接还原为Ride对象——这是 Confluent JSON 序列化器相对原生JsonDeserializer的关键差异必须显式声明类型。消费逻辑采用经典的poll 循环consumer.poll(Duration.of(1, ChronoUnit.SECONDS))每次最多阻塞 1 秒拉取一批消息遍历打印DOLocationID直到连续拉不到消息且循环次数达到 10 次为止。源码中do...while(!results.isEmpty() || i 10)的写法保证了无论有无数据程序都会自然结束适合演示场景。4.3 消息格式与数据模型JSON 示例的数据模型 是公开字段的 POJO如VendorID、DOLocationID、tpep_pickup_datetime、tpep_dropoff_datetime由 Confluent 的KafkaJsonSerializer/KafkaJsonDeserializer负责与 JSON 互相转换。这种无 Schema 强约束的方式上手快但缺少对消息结构的版本化校验这正是下一节 Schema Registry 要解决的问题。五、Schema Registry 与 Avro给消息加上合同5.1 为什么需要 Schema Registry视频 7.12 的主题是 Schema Registry。它的核心价值在于在 Kafka 消息之上再建立一层 Schema 服务生产者注册 Avro Schema消费者按 ID 拉取 Schema 反序列化从而保证消息格式的强类型校验写入前必须符合已注册的 Schema向后兼容的演进新增字段时老消费者不会被打挂Schema 的集中管理所有生产消费者共享同一份定义无需在客户端硬编码。5.2 三种 Avro Schema兼容性演示仓库 src/main/avro 下放了三份 Schema用于演示演进策略基础版 rides.avsc{ type: record, name: RideRecord, namespace: schemaregistry, fields: [ {name: vendor_id, type: string}, {name: passenger_count, type: int}, {name: trip_distance, type: double} ] }兼容演进版 rides_compatible.avsc新增的pu_location_id字段声明为{type: [null, long], default: null}——即可空 带默认值老数据读取新 Schema 时该字段回退为 null属于向后兼容的演进{name: pu_location_id, type: [ null, long ], default: null}第三份rides_non_compatible.avsc则刻意演示非兼容的改动例如删除字段或修改字段类型用于对比说明怎样的 Schema 变更会被 Schema Registry 拒绝。5.3 AvroProducer结合 Schema Registry 的生产者AvroProducer.java 展示了完整接入流程props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName()); props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, https://psrc-kk5gg.europe-west3.gcp.confluent.cloud); props.put(basic.auth.credentials.source, USER_INFO); props.put(basic.auth.user.info, Secrets.SCHEMA_REGISTRY_KEY : Secrets.SCHEMA_REGISTRY_SECRET);新增的四个关键点值序列化器换成KafkaAvroSerializer键仍用 Stringschema.registry.urlSchema Registry 服务地址basic.auth.credentials.source USER_INFO以 HTTP Basic Auth 方式认证 Schema Registrybasic.auth.user.infokey:secret形式的凭据同样取自 Secrets.java。数据侧使用 Avro 生成的强类型类RideRecordRideRecord.newBuilder().setVendorId(...).setTripDistance(...).build()通过RideRecord构造器逐字段赋值后发送到rides_avro主题。注意这里的schemaregistry.RideRecord类正是由 rides.avsc 中的namespace: schemaregistryname: RideRecord生成的——Schema 文件、Java 类、Kafka 消息三者一一对应。另外CustomSerdes.java 中提供getAvroSerde(boolean isKey, String schemaRegistryUrl)工厂方法用SpecificAvroSerde配合 Schema Registry URL 构造 Avro Serde供 Kafka Streams 的拓扑使用这里可以看到 JSON 与 Avro 两条序列化路径在同一个工具类中的统一收口。六、Kafka Streams 实战分组计数与时间窗口6.1 拓扑Topology思维Kafka Streams 的核心编程模型是声明式拓扑用StreamsBuilder描述数据流如何从输入主题经一系列算子map、filter、groupBy、window 等流向输出主题最后由KafkaStreams运行。视频 7.7Kafka stream basics讲解的就是这套模型仓库中每个 KStream 示例类都遵循createTopology()构建 main()启动的结构。6.2 JsonKStream按 Key 分组计数JsonKStream.java 是最简单的流处理示例完整拓扑只有三行StreamsBuilder streamsBuilder new StreamsBuilder(); var ridesStream streamsBuilder.stream(rides, Consumed.with(Serdes.String(), CustomSerdes.getSerde(Ride.class))); var puLocationCount ridesStream.groupByKey().count().toStream(); puLocationCount.to(rides-pulocation-count, Produced.with(Serdes.String(), Serdes.Long())); return streamsBuilder.build();它的语义是从rides主题读取消息Key 是 DOLocationID按 Key 分组并累计计数把(DOLocationID, 计数)结果写入rides-pulocation-count主题。这一模式就是视频 7.7 讲解的KStream 基础的落地实现。启动与关停代码也很有代表性var kStreams new KafkaStreams(topology, props); kStreams.start(); while (kStreams.state() ! KafkaStreams.State.RUNNING) { System.out.println(kStreams.state()); Thread.sleep(1000); } Runtime.getRuntime().addShutdownHook(new Thread(kStreams::close));其中CACHE_MAX_BYTES_BUFFERING_CONFIG 0会关闭记录缓存让聚合结果即时向下游 emit便于课堂观察Runtime.addShutdownHook保证 CtrlC 时优雅关闭流应用。6.3 JsonKStreamWindow10 秒滑动窗口计数JsonKStreamWindow.java 在上一例基础上加入时间窗口对应视频 7.10 Kafka stream windowingvar puLocationCount ridesStream.groupByKey() .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofSeconds(10), Duration.ofSeconds(5))) .count().toStream(); var windowSerde WindowedSerdes.timeWindowedSerdeFrom(String.class, 10 * 1000); puLocationCount.to(rides-pulocation-window-count, Produced.with(windowSerde, Serdes.Long()));TimeWindows.ofSizeAndGrace(10s, 5s)定义 10 秒的窗口大小并允许晚到事件有 5 秒的宽限期grace period输出 Key 从普通 String 变成WindowedString窗口 原始 Key因此必须用WindowedSerdes.timeWindowedSerdeFrom(String.class, 10000)构造窗口 Serde否则无法正确序列化输出主题。结合 4.1 节 Producer 中pickup 时间 现在 - 20 分钟的时间改写不难理解Producer 的静态 CSV 数据必须变成实时时间才能喂给窗口计算这也解释了为什么 JsonProducer.java 要动态改写时间字段。6.4 流处理的状态与键控语义从源码结构看上述两个示例都调用了groupByKey()——这是 Kafka Streams 中把无界流状态化的关键算子相同 Key 的记录被路由到同一个分区、聚合状态按 Key 维护。这也从侧面印证了视频 7.3 中分区内有序的模型只有 Key 相同计数以及后续的 Join才能在单个处理实例上正确合并。七、Kafka Streams Join把两个流关联起来7.1 双流关联的建模视频 7.8Kafka stream join讲解的是 KStream-KStream Join。JsonKStreamJoins.java 给出了完整的出租车场景把rides行程流含下车时间与rides_location上车地点流含上车时间按上车地点 ID 关联生成vendor_info输出。拓扑构建过程分三步定义两个输入流rides反序列化为Riderides_location反序列化为PickupLocation重设 KeypickupLocations.selectKey((key, value) - String.valueOf(value.PULocationID))——Join 要求两流按相同 Key 分区因此把上车地点流按PULocationID重新设 Key与行程流按DOLocationID对齐执行 Joinvar joined rides.join(pickupLocationsKeyedOnPUId, (ride, pickupLocation) - { var period Duration.between(ride.tpep_dropoff_datetime, pickupLocation.tpep_pickup_datetime); if (period.abs().toMinutes() 10) return Optional.empty(); else return Optional.of(new VendorInfo(ride.VendorID, pickupLocation.PULocationID, pickupLocation.tpep_pickup_datetime, ride.tpep_dropoff_datetime)); }, JoinWindows.ofTimeDifferenceAndGrace(Duration.ofMinutes(20), Duration.ofMinutes(5)), StreamJoined.with(Serdes.String(), CustomSerdes.getSerde(Ride.class), CustomSerdes.getSerde(PickupLocation.class)));这里的 ValueJoiner 逻辑值得细读它计算下车时间与上车时间之差若超过 10 分钟则返回Optional.empty()视为不匹配否则构造VendorInfo结果。也就是说即便两流 Key 相同还要通过业务时间约束进一步过滤避免把时间上毫无关联的记录硬拼在一起。7.2 窗口与结果过滤JoinWindows.ofTimeDifferenceAndGrace(20min, 5min)只关联 20 分钟时间差内的记录并允许 5 分钟晚到宽限——时间窗口在这里既是关联边界也是乱序数据的容错机制最终输出前先filter(value - value.isPresent())去掉空结果再mapValues(Optional::get)解包写入Topics.OUTPUT_TOPICvendor_info。7.3 异常处理与运行状态JsonKStreamJoins.java 还额外演示了未捕获异常处理器kStreams.setUncaughtExceptionHandler(exception - { System.out.println(exception.getMessage()); return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION; });当流线程抛出未捕获异常时选择SHUTDOWN_APPLICATION关闭整个应用而非仅重启线程——在演示与教学场景中这种方式能更直观地暴露问题。八、用 TopologyTestDriver 测试流拓扑视频 7.9Kafka stream testing强调流处理代码同样要可测试仓库为此提供了两份 JUnit 5 测试。8.1 计数拓扑测试JsonKStreamTest.java 使用TopologyTestDriver在不连接真实 Kafka 集群的情况下驱动拓扑核心模式是用APPLICATION_ID_CONFIG 假 bootstrapdummy:1234构造测试属性testDriver.createInputTopic(...)与createOutputTopic(...)分别模拟输入/输出主题用inputTopic.pipeInput(key, value)注入消息从outputTopic.readKeyValue()断言输出。三个测试用例覆盖了典型的计数语义单条消息输入一条 → 输出(key, 1L)不同 Key 的两条消息各计各的输出两个1L相同 Key 的两条消息同一 Key 计数累加依次输出1L、2L。8.2 Join 拓扑测试JsonKStreamJoinsTest.java 验证 Join 行为同时向rides与rides_location注入 Key 相同的记录后断言输出主题队列长度为 1且输出VendorInfo的VendorID、pickupTime与期望一致。两份测试共同印证了 Kafka Streams 的一个关键工程属性拓扑逻辑与运行环境解耦——createTopology()返回纯描述既可以挂到生产集群上跑也可以直接交给TopologyTestDriver做离线断言。这也是把业务逻辑与集群连接分离这一最佳实践的生动示范。九、延伸主题ksqlDB 与 Kafka Connect视频 7.11 覆盖了 ksqlDB 与 Kafka Connect。仓库在 07-streaming/extras/ksqldb/commands.md 中提供了配套的 ksqlDB 命令速查演示如何用 SQL 风格语法对 Kafka 主题做流式查询。这一部分属于进阶选学当需要少写 Java 代码、快速用 SQL 探查流数据时ksqlDB 是很轻量的选择而 Kafka Connect 则用于把 Kafka 与外部系统数据库、对象存储等对接。两者与 Kafka Streams 互补共同构成完整的流处理生态。十、如何运行这套示例按照仓库的实际结构运行步骤如下准备集群与凭据创建 Confluent Cloud或兼容 Kafka集群将集群地址替换进各示例类的BOOTSTRAP_SERVERS_CONFIG把 API Key/Secret 填入 Secrets.java使用 Avro 示例时还需创建 Schema Registry 并替换SCHEMA_REGISTRY_URL_CONFIG与 Basic Auth 凭据构建工程在 07-streaming/theory/java/kafka_examples 目录执行./gradlew buildWindows 用gradlew.batGradle Wrapper 会自动下载依赖按顺序运行先跑 JsonProducer.java./gradlew run或在 IDE 中直接运行 main 方法向rides发布消息再跑 JsonConsumer.java 验证消费然后依次运行 JsonKStream.java、JsonKStreamWindow.java、JsonKStreamJoins.java用消费端或 Confluent Cloud 控制台观察输出主题运行单元测试执行./gradlew test不依赖集群即可验证两个拓扑的正确性。前提与限制JSON 与 Avro 示例均面向 Confluent Cloud 环境编写SASL_SSL PLAIN 认证、Schema Registry 的 Basic Auth若使用本地或自建 Kafka需要相应调整安全协议与注册表配置示例中的 bootstrap 地址、Schema Registry 地址均为课程占位示例务必替换。十一、小结理论篇 07-streaming/theory/README.md 用两讲Stream processing 与 Kafka Streams串联起 Kafka 从入门到进阶的完整知识链而 kafka_examples 工程则把每一条理论都变成了可运行的 Java 代码Producer 演示消息发布与 JSON 序列化Consumer 演示消费语义与反序列化AvroProducer 演示 Schema Registry 的强 Schema 治理三个 KStream 类分别覆盖分组计数、时间窗口与流 Join两份测试则展示了不依赖集群的拓扑验证方法。把这些内容与 07-streaming 模块主线Redpanda Python Flink对照学习你既能掌握流处理的通用理论也能看到同一套思想在不同技术栈中的落地形态。【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考