Data Engineering Zoomcamp 2026 模块七：Redpanda + PyFlink 流式作业官方题解实战指南

Data Engineering Zoomcamp 2026 模块七：Redpanda + PyFlink 流式作业官方题解实战指南

【免费下载链接】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 2026 第 7 周（流处理模块）家庭作业的完整题解，基于 solutions.md 展开，并结合仓库中的源码与配置逐行剖析。你将掌握一条端到端的实时数据处理链路：用 Redpanda（Kafka 协议兼容）作为消息中间件、用 Python producer/consumer 完成数据的生产与消费、再借助 PyFlink 的 Table API 以 tumbling 窗口和 session 窗口完成聚合分析并写入 PostgreSQL。读完本文，你能独立复现六道题的完整答案，并理解并行度、水位线（watermark）与窗口触发机制这些流式处理中的关键细节。

背景：作业数据与基础设施

本作业使用 2025 年 10 月的 Green Taxi Trip 数据（parquet 格式），完整背景见 homework.md。基础设施直接复用 workshop 目录下的 Docker 编排，启动命令为：

cd 07-streaming/workshop/ docker compose build docker compose up -d

如需彻底清理旧容器与卷，可先执行docker compose down -v再重新构建。从 docker-compose.yml 可以看到整套环境包含四个服务：

关键细节：容器内 Flink 通过redpanda:29092访问 Kafka 协议端口，宿主机则通过localhost:9092；./src/目录被挂载到 Flink 容器的/opt/src，这正是后续作业文件放置位置的由来。容器名（如workshop-redpanda-1）基于目录名workshop生成，若重命名目录需相应调整命令。

问题 1：确认 Redpanda 版本

进入 Redpanda 容器执行rpk（Redpanda 自带的管理 CLI）查看版本：

docker exec -it workshop-redpanda-1 rpk version

答案：v25.3.9，与 docker-compose.yml 中的镜像标签redpandadata/redpanda:v25.3.9完全一致。这个答案也提示了一个验证技巧：任何容器内工具版本都可以与编排文件中的镜像 tag 交叉核对。

问题 2：向 Redpanda 生产数据

创建 Topic

docker exec -it workshop-redpanda-1 rpk topic create green-trips

生产者实现剖析

仓库中的 producer.py 完整展示了数据生产流程，核心要点如下：

• 读取并裁剪列：用pd.read_parquet(url, columns=[...])直接从云端读取 parquet，仅保留作业要求的 8 个字段（lpep_pickup_datetime、lpep_dropoff_datetime、PULocationID、DOLocationID、passenger_count、trip_distance、tip_amount、total_amount）。
• 时间序列化：datetime 列必须先转成字符串才能 JSON 序列化，使用df['lpep_pickup_datetime'].dt.strftime('%Y-%m-%d %H:%M:%S')。
• 空值处理：passenger_count用fillna(0).astype(int)填充 NaN 并转为整型，避免 JSON 中出现NaN导致下游解析失败。
• 发送与计时：KafkaProducer配置value_serializer=json_serializer（json.dumps(...).encode('utf-8')）；逐行producer.send(topic_name, value=message)后调用producer.flush()确保全部落盘，并用time()记录耗时。

运行方式：

python producer.py

答案：数据集共 49,416 行，发送全程约 10 秒（视机器性能略有浮动）。这也印证了flush()的语义——它阻塞直到缓冲区中的全部消息发送完成，是准确计时的前提。

问题 3：消费者统计 trip_distance > 5 的行程数

consumer.py 给出了一个极简但完整的消费者模板：

consumer = KafkaConsumer( topic_name, bootstrap_servers=[server], auto_offset_reset='earliest', # 从头读取所有消息 group_id='green-trips-homework', value_deserializer=json_deserializer, consumer_timeout_ms=10000, # 10 秒无新消息即停止 )

关键参数语义：

• auto_offset_reset='earliest'：从分区最旧偏移量开始消费，保证统计到全部 49,416 条消息；
• consumer_timeout_ms=10000：消费完数据后 10 秒内没有新消息就自动退出循环，避免程序永久挂起；
• 计数器在循环内对trip['trip_distance'] > 5.0累加。

答案：trip_distance > 5.0的行程共 8506 条（总消息数 49,416 条）。

PyFlink 部分：三个作业的公共骨架

问题 4～6 的三个 Flink 作业结构高度一致，理解公共骨架即可一通百通。以 tumbling_job.py 为例，它由四部分组成：

1. 环境初始化——显式设置并行度与检查点：

env = StreamExecutionEnvironment.get_execution_environment() env.enable_checkpointing(10 * 1000) env.set_parallelism(1)

2. Kafka Source DDL——把字符串时间戳转换为事件时间并声明水位线：

CREATE TABLE green_trips ( lpep_pickup_datetime VARCHAR, ... event_timestamp AS TO_TIMESTAMP(lpep_pickup_datetime, 'yyyy-MM-dd HH:mm:ss'), WATERMARK FOR event_timestamp AS event_timestamp - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'properties.bootstrap.servers' = 'redpanda:29092', 'topic' = 'green-trips', 'scan.startup.mode' = 'earliest-offset', 'properties.auto.offset.reset' = 'earliest', 'format' = 'json' );

计算列（computed column）event_timestamp用TO_TIMESTAMP把lpep_pickup_datetime字符串按'yyyy-MM-dd HH:mm:ss'格式解析成时间戳；WATERMARK FOR ... - INTERVAL '5' SECOND声明 5 秒水位线容忍度，允许迟到 5 秒以内的数据仍能参与窗口计算。

3. JDBC Sink DDL——connector = 'jdbc'，连接jdbc:postgresql://postgres:5432/postgres，用户名密码均为postgres，并用PRIMARY KEY (...) NOT ENFORCED声明主键以支持结果更新（upsert）。

4. 窗口聚合 SQL——见各题详解。

必须注意的两个坑

• 并行度必须为 1：green-tripstopic 只有 1 个分区。若并行度过高，空闲的 consumer 子任务会阻止水位线前进，导致窗口永不触发（详细原因见问题 5）；
• 作业是流式常驻的：提交后让作业运行 1～2 分钟直到结果写入 PostgreSQL，再从 Flink Web UI（http://localhost:8081）取消作业或直接查询结果表。若多次发送过数据，需先删除重建 topic 避免重复：

  docker exec -it workshop-redpanda-1 rpk topic delete green-trips

问题 4：5 分钟 Tumbling 窗口统计各 PULocationID 的行程数

建表

CREATE TABLE tumbling_pickup_counts ( window_start TIMESTAMP(3), PULocationID INT, num_trips BIGINT, PRIMARY KEY (window_start, PULocationID) NOT ENFORCED );

提交作业

将 tumbling_job.py 复制到07-streaming/workshop/src/job/（该目录已挂载到容器/opt/src/job/），然后提交：

docker exec -it workshop-jobmanager-1 \ flink run -py /opt/src/job/tumbling_job.py

窗口聚合 SQL

INSERT INTO tumbling_pickup_counts SELECT window_start, PULocationID, COUNT(*) AS num_trips FROM TABLE( TUMBLE(TABLE green_trips, DESCRIPTOR(event_timestamp), INTERVAL '5' MINUTE) ) GROUP BY window_start, PULocationID;

Tumbling 窗口是固定长度、互不重叠的时间桶（5 分钟一个），DESCRIPTOR(event_timestamp)指明按事件时间对齐窗口边界，COUNT(*)统计每个窗口内每个上车地点的行程数。

查询与答案

SELECT PULocationID, num_trips FROM tumbling_pickup_counts ORDER BY num_trips DESC LIMIT 3;

答案：PULocationID = 74，其最繁忙的 5 分钟窗口内有 15 趟行程。

问题 5：Session 窗口找出最长连续会话

建表

CREATE TABLE session_pickup_counts ( session_start TIMESTAMP(3), session_end TIMESTAMP(3), PULocationID INT, num_trips BIGINT, PRIMARY KEY (session_start, PULocationID) NOT ENFORCED );

提交作业与聚合 SQL

将 session_job.py 复制到07-streaming/workshop/src/job/后提交：

docker exec -it workshop-jobmanager-1 \ flink run -py /opt/src/job/session_job.py

INSERT INTO session_pickup_counts SELECT window_start AS session_start, window_end AS session_end, PULocationID, COUNT(*) AS num_trips FROM TABLE( SESSION(TABLE green_trips_session PARTITION BY PULocationID, DESCRIPTOR(event_timestamp), INTERVAL '5' MINUTE) ) GROUP BY window_start, window_end, PULocationID;

Session 窗口与 Tumbling 窗口的本质区别：它没有固定长度，而是按事件间隙切分——当两个事件之间的时间间隔超过INTERVAL '5' MINUTE（5 分钟 gap）时，前一个会话关闭、新会话开启。PARTITION BY PULocationID表示按上车地点分别维护会话，这样每个地点的活动才能独立成组。

查询与答案

SELECT PULocationID, num_trips, session_start, session_end FROM session_pickup_counts ORDER BY num_trips DESC LIMIT 3;

答案：最长会话包含 81 趟行程（PULocationID = 74，会话发生在 2025-10-08 上午）。

深入理解：为什么并行度必须是 1

solutions 文档特别强调了一个易踩的坑：session 作业必须使用并行度 1。原因在于green-trips只有 1 个分区，若并行度大于 1，其余消费者子任务在分配不到数据时成为"空闲子任务"，而 Flink 的水位线只会在所有输入都越过某阈值时才会整体推进；空闲子任务无法产生新的水位线，导致事件时间停滞，session 窗口永远不会触发关闭。同理，这也解释了为什么问题 4 与问题 6 的作业同样设置env.set_parallelism(1)。

问题 6：1 小时 Tumbling 窗口计算每小时小费总额

建表

CREATE TABLE hourly_tips ( window_start TIMESTAMP(3), total_tips DOUBLE, PRIMARY KEY (window_start) NOT ENFORCED );

提交作业与聚合 SQL

将 tip_job.py 复制到07-streaming/workshop/src/job/后提交：

docker exec -it workshop-jobmanager-1 \ flink run -py /opt/src/job/tip_job.py

INSERT INTO hourly_tips SELECT window_start, SUM(tip_amount) AS total_tips FROM TABLE( TUMBLE(TABLE green_trips_tips, DESCRIPTOR(event_timestamp), INTERVAL '1' HOUR) ) GROUP BY window_start;

与问题 4 相比，这里把窗口粒度从 5 分钟放大到 1 小时，聚合函数从COUNT(*)换成SUM(tip_amount)，且不再按PULocationID分组（作业要求统计全地点每小时的小费总额）。

查询与答案

SELECT window_start, total_tips FROM hourly_tips ORDER BY total_tips DESC LIMIT 3;

答案：小费总额最高的小时是2025-10-16 18:00:00，该小时小费总计约$524.96。

总结：一套可复用的流处理模式

回顾全部六道题，可以提炼出一条完整的实战方法论：

1. 基础设施即代码：用 docker compose 一键拉起 Redpanda + Flink（Job/Task Manager）+ PostgreSQL，版本信息（如 Redpandav25.3.9）可直接从编排文件核对；
2. 生产侧：parquet → pandas 列裁剪 → datetime 转字符串 → JSON 序列化 → KafkaProducer 发送 +flush()计时；
3. 消费侧：auto_offset_reset='earliest'全量消费，consumer_timeout_ms控制退出时机；
4. Flink 侧：统一采用"Kafka Source DDL + 事件时间计算列 + 5 秒水位线 + JDBC Sink DDL + 窗口聚合 SQL"的模板，只需替换窗口类型（TUMBLE/SESSION）、窗口长度（5 分钟/1 小时）与聚合函数（COUNT/SUM）即可复用到其他指标；
5. 并行度纪律：单分区 topic 配合窗口计算时必须set_parallelism(1)，这是水位线能否推进、窗口能否触发的关键前提。

如需查看作业源码的完整实现，可进一步阅读 tumbling_job.py、session_job.py、tip_job.py 与 workshop 侧的 aggregation_job.py（其中包含事件时间、水位线与窗口的对照演示），以及 workshop README 了解完整环境搭建步骤。

【免费下载链接】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