Data Engineering Zoomcamp 2027 第 7 周作业实战：用 Redpanda 与 PyFlink 构建绿色出租车实时流处理管道

Data Engineering Zoomcamp 2027 第 7 周作业实战：用 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 2027 年 cohort 第 7 模块（流处理）的作业展开，核心任务是以 Redpanda（Kafka 兼容消息中间件）为消息枢纽、以 PyFlink 为流处理引擎，对 2025 年 10 月的纽约绿色出租车（Green Taxi）行程数据完成"生产 → 消费 → 窗口聚合"的完整实时链路演练。读完本文，你将掌握 Kafka Producer/Consumer 的 Python 写法、Flink SQL 中 tumbling/session 窗口与 watermark 的实战用法，并能在本地 Docker 环境中独立复现并验证全部 6 道作业题。

作业概览：一条完整的实时流处理链路

本作业建立在 07 模块 workshop 的基础设施之上，目标是复用同一套 Docker 编排，把 workshop 中基于黄色出租车（yellow taxi、tpep_前缀字段、ridestopic）的示例，迁移到绿色出租车（green taxi、lpep_前缀字段、green-tripstopic）数据上，从而检验三个核心技能点：

1. 消息生产与消费：用kafka-python库把 parquet 数据写入 Redpanda topic，再以消费者身份读回并做简单过滤统计；
2. 事件时间与窗口：在 PyFlink 中以字符串形式的时间戳声明事件时间，用 5 秒 watermark 容忍乱序，分别实现 tumbling（滚动）与 session（会话）窗口聚合；
3. 结果落库：通过 Flink JDBC 连接器把聚合结果写入 PostgreSQL 并查询验证。

数据来源为纽约 TLC 公开的绿色出租车行程快照文件green_tripdata_2025-10.parquet（2025 年 10 月整月数据），作业环境由 07-streaming/workshop 目录下的 Docker Compose 提供，作业中引用的完整代码镜像位于 cohorts/2027/07-streaming/code。

环境准备：复用 workshop 的流处理基础设施

作业直接复用 workshop 的环境，无需从零搭建。在仓库根目录下进入 workshop 目录并启动全部服务：

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

build会基于 Dockerfile.flink 构建带 Python 与 PyFlink 的自定义 Flink 镜像（起点是官方flink:2.2.0-scala_2.12-java17，内置 Python 3.12、uv 以及 Kafka / JDBC / PostgreSQL 连接器 JAR）。启动完成后，你会得到 docker-compose.yml 中声明的四个服务：

如果你此前运行过 workshop 并残留了旧容器或数据卷，务必做一次彻底的重置，避免 topic 里的旧消息污染统计结果：

docker compose down -v docker compose build docker compose up -d

注意：容器名（如workshop-redpanda-1）假定目录名是workshop。若你重命名了目录，后续所有docker exec命令中的容器名都要相应调整。

数据准备：Green Taxi 与字段筛选

作业数据来自 NYC TLC 的绿色出租车行程文件（2025 年 10 月），读取 parquet 后只需保留以下 8 个字段：

• lpep_pickup_datetime—— 上车时间（datetime，需转字符串）
• lpep_dropoff_datetime—— 下车时间（datetime，需转字符串）
• PULocationID—— 上车出租车区域 ID
• DOLocationID—— 下车出租车区域 ID
• passenger_count—— 乘客数
• trip_distance—— 行程距离（英里）
• tip_amount—— 小费金额
• total_amount—— 总金额

对比 workshop 使用的 producer.py（只取 5 列、时间戳转 epoch 毫秒），本作业字段更多、且要求把 datetime 列序列化为字符串，以便 Flink 用TO_TIMESTAMP(..., 'yyyy-MM-dd HH:mm:ss')解析——这是本作业与 workshop 最关键的差异之一。

Question 1：确认 Redpanda 版本

Redpanda 自带的 CLI 工具rpk用于管理 broker、topic 与消费组。在 Redpanda 容器内执行：

docker exec -it workshop-redpanda-1 rpk version

输出会给出当前运行的 Redpanda 版本号。这道题的目的是确认环境就绪，同时熟悉rpk这个后续创建/删除 topic 都要用到的工具。

Question 2：Producer——把数据写入 green-trips

先创建作业专用 topic（workshop 中的ridestopic 由 broker 首次使用自动创建，这里显式创建更稳妥）：

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

随后编写 producer：读取 parquet → 筛选 8 列 → 每行转成 dict → 以 JSON 格式发送到green-trips。关键点是把两个 datetime 列先转成'%Y-%m-%d %H:%M:%S'格式字符串，否则 JSON 序列化会失败或产生 Flink 无法解析的格式。整体骨架参照 workshop 的 producer.py，改造如下：

import json import pandas as pd from time import time from kafka import KafkaProducer columns = [ 'lpep_pickup_datetime', 'lpep_dropoff_datetime', 'PULocationID', 'DOLocationID', 'passenger_count', 'trip_distance', 'tip_amount', 'total_amount', ] df = pd.read_parquet('green_tripdata_2025-10.parquet', columns=columns) # datetime -> 字符串，与 Flink 的 'yyyy-MM-dd HH:mm:ss' 格式对齐 df['lpep_pickup_datetime'] = df['lpep_pickup_datetime'].dt.strftime('%Y-%m-%d %H:%M:%S') df['lpep_dropoff_datetime'] = df['lpep_dropoff_datetime'].dt.strftime('%Y-%m-%d %H:%M:%S') def json_serializer(data): return json.dumps(data).encode('utf-8') producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=json_serializer, ) t0 = time() for _, row in df.iterrows(): producer.send('green-trips', value=row.to_dict()) producer.flush() t1 = time() print(f'took {(t1 - t0):.2f} seconds')

workshop 里发送 1000 行大约耗时 10 秒（producer.py 中可见time.sleep(0.01)的节流逻辑）；本次是整月数据、全量发送，用时量级取决于网络与机器。作业给出的候选答案是 10 / 60 / 120 / 300 秒，用于判断你测得的时间落在哪个档位。

Question 3：Consumer——统计 trip_distance > 5 的行程数

写一个 Kafka 消费者读回全部消息。必须设置auto_offset_reset='earliest'，否则新消费者默认从latest开始、只会看到订阅之后的新消息，读不到已经写入的历史数据。group_id用于让 Kafka 记录该消费组的读取进度。参考 consumer.py 的写法：

import json from kafka import KafkaConsumer def json_deserializer(data): return json.loads(data.decode('utf-8')) consumer = KafkaConsumer( 'green-trips', bootstrap_servers=['localhost:9092'], auto_offset_reset='earliest', group_id='green-trips-consumer', value_deserializer=json_deserializer, ) count = 0 for message in consumer: if message.value['trip_distance'] > 5.0: count += 1 print(f'trips with trip_distance > 5: {count}')

作业的候选答案是 6506 / 7506 / 8506 / 9506，取最接近你实际统计结果的选项。注意trip_distance在原始数据中为英里，这里直接按数值比较，不涉及单位换算。

Part 2 预备：PyFlink 作业的三个关键差异

进入 PyFlink 部分前，先明确与 workshop 代码（aggregation_job.py）的三处不同：

1. topic 名：green-trips（替代 workshop 的rides）；
2. 字段前缀：datetime 列是lpep_前缀（替代tpep_）；
3. 时间戳形态：本作业的时间是字符串（如2025-10-01 00:00:00），而 workshop 中是 epoch 毫秒整数。因此源表 DDL 中要先用计算列把字符串转成 Flink 时间戳，并声明 watermark：

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

TO_TIMESTAMP按给定格式解析字符串；watermark 取"最新事件时间 − 5 秒"，作为窗口发布结果的触发器，容忍最多 5 秒的乱序到达（workshop 的 README 对此有完整图解：07-streaming/workshop/README.md）。

运行 Flink 作业前还需注意以下几点：

• 结果表先行：在 PostgreSQL 中预先创建好各题的结果表（用docker compose exec postgres psql -U postgres -d postgres或任意 SQL 客户端连接localhost:5432）；
• 作业文件位置：把你的.py作业文件放在workshop/src/job/下，该目录被挂载进 Flink 容器，容器内路径为/opt/src/job/；
• 提交命令：

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

• 并行度必须设为 1：green-trips只有 1 个分区，因此作业里要env.set_parallelism(1)。若并行度过高，没有分配到数据的分区子任务（idle consumer subtask）不会消费消息，会导致 watermark 无法推进、窗口迟迟不发布结果；
• 流式作业常驻运行：Flink 流作业不会自行结束，让作业跑 1～2 分钟、看到 PostgreSQL 里出现结果后，可在 http://localhost:8081 的 Flink UI 上取消作业；
• 重复数据处理：如果 producer 跑过多次，topic 里会有重复消息。想彻底清空可删除并重建 topic：

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

Question 4：5 分钟 Tumbling Window 统计热门接客区

需求：统计每个PULocationID在每个 5 分钟滚动窗口内的行程数。先建结果表（包含window_start、PULocationID、num_trips三列），并建议像 workshop 的聚合表那样加上主键，以启用 upsert 语义、让迟到数据能修正已发布的结果（参见 aggregation_job.py 中PRIMARY KEY (...) NOT ENFORCED的用法）：

CREATE TABLE trips_by_pulocation ( window_start TIMESTAMP, PULocationID INTEGER, num_trips BIGINT, PRIMARY KEY (window_start, PULocationID) );

Flink 作业的核心 SQL 用TUMBLE函数：第一个参数是源表，第二个参数DESCRIPTOR(event_timestamp)必须指向定义了 watermark 的那一列，第三个参数是窗口大小：

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

处理完所有数据后查询：

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

取num_trips最大的PULocationID，候选答案是 42 / 74 / 75 / 166。workshop 用 1 小时窗口做了同样的聚合演示，只是多加了SUM(total_amount)与env.set_parallelism(3)（见 aggregation_job.py），你可以对照该文件理解 TUMBLE 的完整调用形态。

Question 5：Session Window 找最长会话

需求：以lpep_pickup_datetime作为事件时间、5 秒 watermark 容忍，按PULocationID分组做会话窗口（gap 为 5 分钟）。会话窗口与 tumbling 的本质区别在于：窗口大小不固定，事件与上一事件间隔不超过 5 分钟就并入同一会话，一旦出现超过 5 分钟的空档，当前会话关闭。

|--events--| gap(>5min) |--events------| gap(>5min) |--events--| | Session 1| | Session 2 | | Session 3|

Flink SQL 中会话窗口写作SESSION(...)，第一个参数仍是要按事件时间开窗的表：

INSERT INTO longest_sessions SELECT PULocationID, COUNT(*) AS num_trips FROM TABLE( SESSION(TABLE green_trips_source, DESCRIPTOR(event_timestamp), INTERVAL '5' MINUTE) ) GROUP BY PULocationID, window_start, window_end;

结果表自行定义（至少包含PULocationID与num_trips），然后找到单次会话内行程数最多的PULocationID，即"最长会话"。候选答案是 12 / 31 / 51 / 81。workshop README 的"Understanding window types"一节对比了 tumbling / sliding / session 三种窗口的适用场景，会话窗口常用于用户行为归因（07-streaming/workshop/README.md），这里把它用在出租车接客上，语义同样成立。

Question 6：1 小时 Tumbling Window 统计小费高峰

需求：用1 小时 tumbling 窗口聚合全量区域每小时的小费总额SUM(tip_amount)，找出小费最高的那个小时。先建表：

CREATE TABLE hourly_tip_amounts ( window_start TIMESTAMP, total_tip_amount DOUBLE PRECISION );

作业 SQL 与 Q4 同构，只是窗口变为 1 小时、聚合函数换成SUM、且不再按PULocationID分组（跨所有区域汇总）：

INSERT INTO hourly_tip_amounts SELECT window_start, SUM(tip_amount) AS total_tip_amount FROM TABLE( TUMBLE(TABLE green_trips_source, DESCRIPTOR(event_timestamp), INTERVAL '1' HOUR) ) GROUP BY window_start;

查询total_tip_amount最大的window_start，候选答案是2025-10-01 18:00:00/2025-10-16 18:00:00/2025-10-22 08:00:00/2025-10-30 16:00:00。因为这里的结果表没有主键、且每小时只输出一行，不会产生重复更新，可以不用 upsert；不过若担心同一窗口被重复发布，仍可像 Q4 那样加主键。

提交与检查清单

完成全部 6 题后，通过 cohort 提供的作业提交表单提交答案（homework.yaml中声明了提交表单配置与截止时间，见 cohorts/2027/07-streaming/homework.yaml）。提交前建议对照以下清单自检：

• Redpanda 版本号是否来自rpk version的真实输出；
• producer 发送耗时与题面档位一致，且flush()已调用（未 flush 会导致部分消息滞留缓冲区、统计偏少）；
• consumer 统计用的是earliest偏移，且只跑了一次（重复运行同一group_id会从上一次位置继续，导致计数不完整）；
• Flink 作业都设置了env.set_parallelism(1)，watermark 能正常推进，窗口结果已出现在 PostgreSQL 中再取消作业；
• 若 producer 重复发送过数据，已用rpk topic delete green-trips清空后重新发送。

与 workshop 源码的对照学习路径

本作业的所有技术细节都能在仓库的 workshop 源码中找到对应实现，建议按以下顺序对照研读：

• 07-streaming/workshop/docker-compose.yml：四个服务的完整编排，含 Redpanda 双监听地址、Flink 内存与并行度参数；
• 07-streaming/workshop/Dockerfile.flink：PyFlink 镜像构建过程与连接器 JAR 清单；
• 07-streaming/workshop/src/models.py：Ride 数据模型与序列化/反序列化函数，可据此改造成 green taxi 的 8 字段版本；
• 07-streaming/workshop/src/producers/producer.py 与 producer_realtime.py：批量与实时两种生产方式，后者还模拟了 ~20% 的 3～10 秒延迟事件，用于观察 watermark 与 upsert 行为；
• 07-streaming/workshop/src/consumers/consumer.py：earliest偏移消费与反序列化；
• 07-streaming/workshop/src/job/pass_through_job.py：最简 Kafka→JDBC 透传作业，含latest-offset与TO_TIMESTAMP_LTZ的用法；
• 07-streaming/workshop/src/job/aggregation_job.py：TUMBLE 窗口 + watermark + upsert 主键的完整聚合作业，是 Q4/Q6 的直接蓝本。

把 workshop 中ridestopic、tpep_字段、epoch 毫秒时间戳这三处替换成本作业要求的green-tripstopic、lpep_字段与字符串时间戳，再按上述要点调整并行度与结果表，即可系统性地完成全部作业——这也正是"以流处理思维迁移数据形态"的实战训练目标。

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