Data Engineering Zoomcamp 2026 实战用 dlt MCP 从零构建纽约出租车数据管道【免费下载链接】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 dlt 工作坊的课后作业对应仓库文档 cohorts/2026/workshops/dlt/dlt_homework.md为主线完整讲解如何从一个没有现成 dlt 脚手架的自定义 REST API 出发借助 dltdata load tool与 dlt MCP Server 构建生产级数据管道将分页 JSON 数据加载到本地 DuckDB并通过 dlt Dashboard、dlt MCP 对话与 marimo Notebook 三种方式完成数据探索与作业问题解答。读完本文你将掌握 dlt 项目初始化、rest_api_source配置、分页处理、AI 辅助开发工作流以及多种数据检视方法可将其直接复用到你自己的 API 数据管道项目中。一、作业背景与目标本次作业源于 Data Engineering Zoomcamp 2026 的 dlt 工作坊。工作坊主体见 workshops/dlt/README.md演示了如何基于 dlt 官方脚手架scaffold快速搭建 Open Library API 管道并利用 AI 辅助 IDE 生成、调试和运行管道。而本次作业将场景升级为从零构建数据源是一个没有现成 dlt 脚手架的自定义 API你必须自行提供 API 元数据并驱动 AI 助手生成管道代码。作业的核心目标构建一个可运行的 dlt 管道从自定义 API 提取纽约出租车行程数据将数据加载到本地 DuckDB 数据库无需任何云凭据或配置通过 dlt Dashboard、dlt MCP Server 与 marimo Notebook 三种方式探索加载后的数据回答基于数据的 3 个分析问题。本次作业的独特价值在于它不仅训练传统的管道构建技能还引入了AI 辅助开发AI-assisted development工作流——dlt MCP Server 让 AI 助手能够访问 dlt 官方文档、代码示例与管道元数据从而实现用自然语言描述 API → 生成管道代码 → 自动运行加载的端到端体验。二、数据源NYC Yellow Taxi 自定义 API作业使用的数据源是NYC 黄色出租车行程数据通过一个自定义 REST API 提供。该 API 的特点如下表属性值基础 URLhttps://us-central1-dlthub-analytics.cloudfunctions.net/data_engineering_zoomcamp_api数据格式分页 JSONPaginated JSON每页大小1,000 条记录/页分页终止条件当返回空页时停止这个 API 是本次作业的关键难点没有现成脚手架dlt init只提供 dlt 项目骨架不会生成 API 元数据 YAML 文件因此 API 的 URL、分页方式等全部需要你或你的 AI 助手自行配置。分页必须正确处理API 按每页 1,000 条返回分页终止条件是返回空页即当某页没有任何记录时表示数据已全部拉取完毕。如果分页配置错误会导致数据不完整或无限循环。数据内容纽约出租车行程记录包含行程时间、支付方式、金额、小费金额等字段用于回答作业中的三个分析问题。三、准备工作环境与工具链在开始构建管道之前需要准备以下环境与工作坊 README 保持一致见 workshops/dlt/README.md1. Python 3.11python --version # 应显示 3.11 或更高版本2. uv 或 pip工作坊推荐使用 uv 管理依赖curl -LsSf https://astral.sh/uv/install.sh | sh3. Agentic IDE需要一个具备 AI 辅助能力的代码编辑器工作坊推荐IDE说明CursorVS Code 分支内置 AI 辅助工作坊推荐Windsurf备选 agentic IDEVS Code GitHub Copilot可用但集成度略低4. 理解 dlt 基础概念可选但推荐如果你对 dlt 还不熟悉工作坊提供了概念讲解 Notebook对应仓库文件 dlt_Pipeline_Overview.ipynb其中核心概念如下Source数据源管道中负责从某处获取数据的部分在 dlt 中通常用rest_api_source以简单字典配置描述 API而非手写大量请求代码Pipeline管道描述数据的目标位置如 DuckDB并跟踪表、schema 与运行历史Extract → Normalize → Load 三阶段这是 dlt 管道的完整流程详见下文第五节。四、环境配置搭建 dlt 开发环境Step 1创建项目文件夹如果你在工作坊演示中已经创建过项目文件夹如 Open Library 演示项目可以直接复用否则新建mkdir taxi-pipeline cd taxi-pipeline然后在 Cursor或你偏好的 agentic IDE中打开该文件夹。Step 2配置 dlt MCP Server如未配置dlt MCP Server 是本次 AI 辅助开发的核心组件。它让 AI 助手能够访问 dlt 文档、代码示例以及你的管道元数据pipeline metadata从而更准确地生成和调试代码。Cursor 配置方式进入Settings → Tools MCP → New MCP Server添加{ mcpServers: { dlt: { command: uv, args: [ run, --with, dlt[duckdb], --with, dlt-mcp[search], python, -m, dlt_mcp ] } } }VS CodeCopilot配置方式在项目文件夹中创建.vscode/mcp.json{ servers: { dlt: { command: uv, args: [ run, --with, dlt[duckdb], --with, dlt-mcp[search], python, -m, dlt_mcp ] } } }Claude Code 配置方式在终端中运行claude mcp add dlt -- uv run --with dlt[duckdb] --with dlt-mcp[search] python -m dlt_mcp配置要点说明三条命令本质相同——使用uv run临时创建运行环境安装dlt[duckdb]dlt 主库 DuckDB 目标支持与dlt-mcp[search]MCP Server 及其文档搜索能力然后以 Python 模块方式启动dlt_mcp。启动后AI 助手即可通过 MCP 协议查询 dlt 文档、代码示例与当前管道元数据。Step 3安装 dltpip install dlt[workspace]dlt[workspace]是一个聚合安装包含 dlt 核心、常用目标数据库支持以及 MCP Server 等 AI 辅助开发工具。Step 4初始化项目dlt init dlthub:taxi_pipeline duckdb该命令会创建 dlt 项目文件如.dlt/配置目录、requirements.txt等创建用于 AI 辅助的 Cursor 规则Cursor rules但不会创建 API 元数据 YAML 文件——因为dlthub:taxi_pipeline没有对应的脚手架。这也是本次作业与工作坊演示Open Library 有脚手架的核心区别你需要在下一步自行提供 API 信息。五、从零构建管道Extract → Normalize → Load由于该 API 没有脚手架你需要将 API 详细信息写进提示词prompt让 AI 助手生成管道代码。Step 5用提示词驱动 Agent 生成管道以下是作业给出的示例提示词可复制到 Cursor 等 agentic IDE 的对话中Build a REST API source for NYC taxi data. API details: - Base URL: https://us-central1-dlthub-analytics.cloudfunctions.net/data_engineering_zoomcamp_api - Data format: Paginated JSON (1,000 records per page) - Pagination: Stop when an empty page is returned Place the code in taxi_pipeline.py and name the pipeline taxi_pipeline. Use dlt rest api as a tutorial.提示词的关键要素明确说明是自定义 REST API无脚手架可用提供完整 API 元数据Base URL、数据格式分页 JSON、分页规则每页 1,000 条、空页终止指定输出文件与管道命名taxi_pipeline.py、管道名taxi_pipeline引用dlt rest api教程让 AI 采用 dlt 官方的 REST API 最佳实践。理解 Agent 将生成的代码为了让你知道 agent 会生成什么、并能验证其正确性这里给出工作坊演示中 Open Library 管道的完整代码见仓库 open_library_pipeline.py它展示了 dlt REST API 源的标准写法Pipeline to ingest data from the Open Library Search API. import dlt from dlt.sources.rest_api import rest_api_source def open_library_source(query: str harry potter): Create a dlt source for the Open Library Search API. Args: query: Search query string (default: harry potter) return rest_api_source({ client: { base_url: https://openlibrary.org, }, resource_defaults: { primary_key: key, write_disposition: replace, }, resources: [ { name: books, endpoint: { path: search.json, params: { q: query, limit: 100, }, data_selector: docs, paginator: { type: offset, limit: 100, offset_param: offset, limit_param: limit, total_path: numFound, }, }, }, ], }) if __name__ __main__: pipeline dlt.pipeline( pipeline_nameopen_library_pipeline, destinationduckdb, dataset_nameopen_library_data, progresslog, ) # Load Harry Potter books from Open Library load_info pipeline.run(open_library_source(queryharry potter)) print(load_info)对照该示例你的出租车管道在生成时应重点关注以下几个配置点配置项作用出租车作业中的注意点client.base_urlAPI 基础 URL应替换为作业给定的自定义 API URLresources[].name资源名称决定生成表名endpoint.pathAPI 路径自定义 API 通常为根路径或固定路径data_selector从 JSON 响应中选取数据数组需根据 API 返回结构确定如docs、results或根数组paginator分页器配置作业 API 为空页终止分页可使用 dlt 的分页器处理Open Library 示例用的是offset分页逻辑可参考但不完全相同write_disposition写入策略replace表示每次全量替换适合此类作业场景关于分页的关键提示作业 API 的分页终止条件是返回空页stop when an empty page is returned。dlt 的 REST API 源内置多种分页器offset、cursor、page number 等agent 会为自定义 API 选择合适的配置。如果管道运行后加载的行数明显偏少如只有 1,000 条说明分页可能只拉取了第一页需要检查 paginator 配置。Step 6运行与调试生成代码后运行管道python taxi_pipeline.py如果出现错误将错误信息粘贴到对话中让 agent 调试。这是 AI 辅助开发的典型迭代流程运行 → 报错 → 反馈给 AI → 修复 → 重跑。管道运行成功后dlt 会依次执行三个阶段Extract提取向自定义 API 发送请求下载原始 JSON 响应存放到 dlt 本地工作目录。此时数据尚未进入 DuckDB。Normalize归一化将嵌套 JSON 转换为关系型表结构。dlt 会为每个表添加_dlt_id行唯一标识与_dlt_load_id关联加载任务跟踪列将嵌套列表展开为子表如trips__xxx形式的子表通过_dlt_parent_id关联父表并创建_dlt_loads、_dlt_pipeline_state、_dlt_version等元数据表。Load加载在 DuckDB 中创建表若不存在并插入归一化后的数据同时记录加载历史。在完整理解这三阶段后也可以直接用pipeline.run(source)一条命令完成全部流程它等价于extract → normalize → load三步。六、探索加载后的数据三种方法管道运行成功后作业要求使用工作坊中讲解的方法来调查数据。仓库中的分析示例见 analysis.py展示了 marimo 与 ibis 的组合用法可作参考。方法一dlt Dashboarddlt pipeline taxi_pipeline show这会启动一个 Web 应用用于查看管道状态与运行历史浏览 schema、表与列结构查询已加载数据调试可能存在的问题。方法二dlt MCP Server 对话式查询配置好 dlt MCP Server 后可以直接在对话中向 AI 提问例如What tables were created in the pipeline? Show me the schema for the trips table. How many rows were loaded?agent 可以访问你的管道元数据直接回答这些问题无需手写 SQL。方法三marimo Notebook 可视化分析创建 marimo notebook 进行查询与可视化。工作坊的推荐运行方式marimo edit your_notebook.py # 编辑模式开发 marimo run your_notebook.py # 运行模式查看报告工作坊的示例 notebookanalysis.py展示了 dlt marimo ibis 的标准分析流程import marimo as mo import dlt import ibis import altair as alt from dlt.helpers.marimo import render, load_package_viewer # 使用 dlt 原生接口访问管道与数据集 pipeline dlt.attach(open_library_pipeline) dataset pipeline.dataset() # 获取 ibis 连接进行丰富的数据探索 ibis_con dataset.ibis()核心步骤使用dlt.attach(pipeline_name)重新挂载已存在的管道通过pipeline.dataset()获取数据集接口无需手写 SQL使用dataset.ibis()获得 ibis 连接进行链式查询group_by、agg、order_by 等使用 Altair 绘制图表通过render(load_package_viewer)直接在 notebook 中嵌入 dlt 包查看器。依赖清单工作坊项目的依赖配置见 pyproject.toml可作为环境参考dependencies [ altair6.0.0, dlt[workspace]1.21.0, ibis-framework[duckdb]12.0.0, jupyterlab4.5.4, marimo0.19.9, ]七、作业问题与解题思路管道成功运行后请基于加载到 DuckDB 的数据回答以下三个问题Question 1数据集的开始日期和结束日期是什么2009-01-01 至 2009-01-312009-06-01 至 2009-07-012024-01-01 至 2024-02-012024-06-01 至 2024-07-01解题思路对行程时间字段执行 MIN/MAX 聚合。注意行程时间字段可能是字符串格式必要时使用 CAST 转换为日期类型后再取极值。Question 2使用信用卡支付的行程占比是多少16.66%26.66%36.66%46.66%解题思路支付方式字段如payment_type中识别信用卡对应的枚举值计算信用卡行程数 / 总行程数。建议用 GROUP BY 先查看该字段的取值分布确认信用卡的取值后再计算比例。Question 3小费tips产生的总金额是多少$4,063.41$6,063.41$8,063.41$10,063.41解题思路对小费金额字段执行 SUM 聚合。同样注意字段的数值类型转换。提交与截止日期通过课程官网的作业提交表单提交答案对应courses.datatalks.club/de-zoomcamp-2026/homework/dlt注意截止时间以网站公布为准。八、实用技巧与常见问题作业文档给出了几条关键提示值得展开说明API 返回分页数据确保管道正确处理分页这是最容易出错的地方。验证方法对比加载的总行数与 API 返回的总记录数。若只有 1 页数据被加载1,000 行说明分页器配置不正确。由于终止条件是空页要确认 agent 生成的分页器在遇到空页时能正常结束循环而不是报错或死循环。Agent 卡住时把错误信息粘贴到对话中让它调试AI 辅助开发的核心理念就是出错即反馈。错误信息traceback是 agent 调试的最重要线索粘贴完整报错让它修复后重跑。使用 dlt MCP Server 查询管道元数据如加载了多少行、建了哪些表这类问题直接问 agent 即可无需手写 SQL。这既能验证管道正确性也能练习 MCP 工作流。尝试多种调查方法作业鼓励在回答问题时尝试 dlt Dashboard、MCP 对话、marimo 可视化等不同方法并分享哪种方法效果最好——这也是本次作业的隐性训练目标。九、参考资料资源说明workshops/dlt/README.md工作坊完整指南Open Library 演示管道dlt_Pipeline_Overview.ipynbdlt 概念讲解 NotebookExtract/Normalize/Loadopen_library_pipeline.py工作坊演示管道的完整代码REST API 源标准写法analysis.pymarimo ibis 数据探索示例pyproject.toml工作坊依赖清单images/etl_diagram.pngExtract → Normalize → Load 流程示意图十、学习在公开中分享Learning in Public课程鼓励所有学员公开分享学习成果learning in public。作业文档提供了 LinkedIn 与 Twitter/X 的发帖模板其要点如下总结你在工作坊中学到的技能REST API 数据管道、dlt MCP Server 的 AI 辅助开发、分页 API 数据加载到 DuckDB、dlt Dashboard 与 marimo 数据检视分享你的作业解决方案链接介绍 Data Engineering Zoomcamp 这一免费课程。分享不仅是社区文化的体现也是数据工程师建立个人品牌、沉淀知识体系的有效方式。【免费下载链接】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），仅供参考
