@freeCodeCamp: 生产级ETL管道需要保持可靠运行，即使数据混乱或API失败。在本手册中，Brookly…

生产级ETL管道需要持续可靠地运行，即使数据混乱或API失败。在本手册中，Brooklyn将向你展示如何使用真实的洪水数据，用Python构建一条生产级ETL管道。你将学习增量加载、类型转换、去重、幂等性，以及构建可靠数据管道的其他模式。https://freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/… — # ETL管道手册：如何用Python构建生产级管道 来源：https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/ ETL管道手册：如何用Python构建生产级管道 追踪洪水风险需要的不是光鲜亮丽的东西，而是一件至关重要的事情：干净且结构化的数据。在本教程中，你将亲手构建一条数据管道。你将创建一个Python ETL（提取、转换、加载）管道，从Hub’Eau（https://hubeau.eaufrance.fr/）——法国官方的开放水文数据API——拉取每日水位读数。然后，你将清洗这些数据，并将其发布为公开数据集，就像实时版本（https://www.kaggle.com/code/grimespoint/paris-flood-dataset-weekly-updater）所做的那样。本教程基于一条每周按计划运行一次的真实管道，它自动保持巴黎洪水数据集（https://www.kaggle.com/datasets/grimespoint/paris-flood-dataset）的更新。不过，你不仅仅是复制粘贴代码。真正的目标是理解管道为何以这种方式工作。你将逐步了解那些设计决策——正是这些决策让一个“只运行一次”的脚本与一个能在无人看管的情况下持续工作多年的脚本区别开来。你可以跟着这个notebook（https://www.kaggle.com/code/grimespoint/data-engineering-with-python-etl-pipeline）一起编码。在本教程的大部分内容中，管道运行在模拟（mock）API数据上。这样你可以安全地运行每个单元格，而不会向真实服务器发送大量请求。稍后的一节将展示如何切换到实时API。到本教程结束时，你将能够： - 解释并实现提取、转换、加载模式 - 使用Python@dataclass管理配置，而不是将常量散落在各处 - 编写能够应对网络故障和分页响应的API抓取代码 - 应用健壮的类型转换，使一行坏数据不会导致整个管道运行崩溃 - 安全地去重和合并增量数据 - 将所有内容整合到单一、幂等、可调度的main()入口点 ### 目录： - 先决条件 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-prerequisites) - 第1部分：设计逻辑 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-1-the-design-logic) - 什么是ETL管道？ (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-what-is-an-etl-pipeline) - 架构总览 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-the-architecture-at-a-glance) - 使其达到生产级标准的两种模式 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-two-patterns-that-make-it-production-grade) - 第2部分：设置与依赖 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-2-setup-and-dependencies) - 第3部分：使用数据类管理配置 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-3-manage-configuration-with-dataclasses) - 为什么需要一个配置层？ (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-why-bother-with-a-config-layer-at-all) - 代码级讲解 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-code-level-walkthrough) - 第4部分：提取步骤 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-4-the-extraction-step) - 优雅地加载文件 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-graceful-file-loading) - 增量更新逻辑 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-incremental-update-logic) - 模拟API (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-simulate-the-api) - 真实抓取数据 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-fetch-data-for-real) - 第5部分：转换步骤 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-5-the-transform-step) - 类型解析与优雅转换 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-type-parsing-and-graceful-coercion) - 通过双向映射进行模式翻译 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-schema-translation-with-bidirectional-mappings) - 计算洪水警报 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-compute-flood-alerts) - 列排序 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-column-ordering) - 去重 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-deduplication) - 在postprocess()中整合所有步骤 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-put-it-all-together-in-postprocess) - 第6部分：加载步骤 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-6-the-load-step) - 设计逻辑 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-design-logic) - 代码级讲解 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-code-level-walkthrough) - 第7部分：组装完整管道 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-7-assemble-the-full-pipeline) - 全局演练（模拟模式） (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-the-global-rehearsal-mock-mode) - main()：管道编排 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-main-pipeline-orchestration) - if name == "main":保护 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-the-if-name-main-guard) - 第8部分：上线并切换到真实API (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-part-8-go-live-and-switch-to-the-real-api) - 运行后验证 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-post-run-validation) - 总结 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-summary) - 参考资料 (https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-references) ## 先决条件 - Python 3.10+。代码使用了类型提示和数据类。这些从Python 3.7开始就能工作，但3.10+最好。 - 掌握pandas DataFrame（https://leetcode.com/studyplan/introduction-to-pandas/）：读取数据、过滤行、基本列操作。 - 熟悉Python中的函数和基本OOP（面向对象编程）（https://realpython.com/python3-object-oriented-programming/）。别担心，本指南会在你学习过程中解释每一处不明显的代码。 - 可选：一个免费的Kaggle（https://www.kaggle.com/）账户和Kaggle CLI（https://github.com/Kaggle/kaggle-api），仅在你想要真实运行最终发布步骤时需要。 安装依赖： pip install requests pandas numpy ipykernel 你将在Jupyter notebook（https://www.kaggle.com/code/grimespoint/data-engineering-with-python-etl-pipeline）中完成工作。从Kaggle下载.ipynb文件，或点击“Copy and Edit”直接在Kaggle上操作。你需要一个Kaggle账户才能这样做。菜单按钮提供下载数据工程跟练notebook的选项。可选：如果你计划在本地运行notebook，请同时安装notebook包： pip install notebook ## 第1部分：设计逻辑 在你写任何代码之前，我们先花三分钟时间看看管道为什么是这样的形态。这是宏观视角。后面每一个代码级决策都可以追溯到这些想法之一。即使你略读其他所有内容，也请阅读本节。 ### 什么是ETL管道？ ETL代表提取、转换、加载。这是以可靠、可重复的方式将数据从源移动到目标的标准模式。 - 提取：从API、数据库或文件等源拉取数据。 - 转换：清洗、标准化、丰富和验证数据。 - 加载：将结果写入目标，如数据仓库、CSV或公共平台。 以下是这三个阶段如何映射到这个项目上： 阶段这里发生了什么提取如果存在现有数据则加载它，然后通过requests为每个水文站调用Hub’Eau API转换将法语列和条目翻译成英语，修复数据类型，删除重复项加载写入CSV和元数据文件，然后通过CLI发布到Kaggle### 架构总览 一个简单ETL管道的示意图。注意，“提取”已经触及了两个不同的源：现有数据集（你已经拥有的）和来自API的新数据。这一区别是下一个想法的种子。 ### 使其达到生产级标准的两种模式 有两种模式使这条管道在无人值守、按计划运行多年时依然安全。 #### 1. 幂等性 幂等性（https://en.wikipedia.org/wiki/Idempotence）意味着同一操作运行两次与运行一次产生相同的结果。去重使这条管道具有幂等性。如果调度器意外触发两次，或者网络重试再次抓取同一天的数据，第二次运行不会创建重复行。这对于任何按计划运行且无人监控的事情都非常重要。 #### 2. 增量加载 一个天真的管道每次运行都会重新下载整个历史数据。这很慢、浪费API配额，而且脆弱：传输的数据越多，失败的可能性就越大。增量管道避免了这一点。相反，它： - 检查数据集中已存在的最新日期 - 只请求从该日期起的数据 - 将新记录合并到现有数据集中 你会看到这个逻辑明确地构建在determine_update_range（https://www.freecodecamp.org/news/the-etl-pipeline-handbook-how-to-build-a-production-grade-pipeline-in-python/#heading-incremental-update-logic）中。永远问自己：“我真的需要这样做吗？”这个习惯将脆弱脚本与生产级管道区分开来。例如，这条管道中的每一个“昂贵”或“外部”步骤（网络调用、磁盘写入、发布）都由便宜的本地检查先行保护。 ## 第2部分：设置与依赖 这是导入块。它看起来平平无奇，但它的组织方式本身就是一种值得指出的最佳实践。 # 标准库导入 import json import os import random import subprocess from dataclasses import dataclass, field from datetime import date, timedelta from pathlib import Path from typing import Dict, List, Optional, Set, Tuple # 第三方库 import numpy as np import pandas as pd import requests # Notebook显示配置 pd.set_option('display.max_columns', None) # 显示所有列 pd.set_option('display.max_colwidth', None) # 防止截断长列文本并显示... 代码级说明： - 导入分为三个块：标准库、第三方、本地，块之间用空行隔开。这遵循PEP 8的导入顺序约定（https://peps.python.org/pep-0008/#imports）。大多数自动格式化工具（https://en.wikipedia.org/wiki/Pretty-printing）（isort、ruff）都强制执行这种分组。 - 没有使用from module import *导入任何内容。该语法会污染当前命名空间。当六个月后有人调试代码时，很难追踪某个名字来自哪里。Python风格指南也表达了同样的建议，包括Google Python风格指南（https://google.github.io/styleguide/pyguide.html）。 - pd.set_option(...)调用纯粹是为了notebook的可读性，这样宽DataFrame不会被...截断。它们对管道的逻辑没有影响。在.py脚本中，你通常会移除它们或以不同方式限定范围。 ## 第3部分：使用数据类管理配置 ### 为什么需要一个配置层？ 每条管道都有旋钮：要监控哪些站点、洪水阈值是多少、发布到哪里。糟糕的做法是在编写代码时将这些值作为字面量散落在代码中。最终你会得到一个埋在三层函数深处的if level > 6000:。然后更改任何设置都意味着在整个文件中寻找，并且很容易更新一处而漏掉另一处。解决方案：将所有设置集中在一个地方。Python的@dataclass（https://docs.python.org/3/library/dataclasses.html）装饰器天然适合这一点。 特性为何重要自动生成的__init__、__repr__、__eq__你不需要自己编写类型提示（https://www.geeksforgeeks.org/python/type-hints-in-python/）给你IDE自动补全和自我文档化代码可选的frozen=True给你真正的不可变性（https://stackoverflow.com/questions/66194804/what-does-frozen-mean-for-dataclasses），如果你希望配置旋钮在创建后不能被更改__post_init__钩子在构造后立即验证或计算派生字段一次将其与纯dict编写的配置进行比较： config = { "stations": ["STN001", "STN002"], "flood_threshold": 6000, "publish_url": "https://example.com/alerts", "retry_count": 2, "timeout_seconds": 5, } 纯dict不会给你这些：没有不可变性、没有类型检查、也没有自动补全。 ### 代码级讲解 管道定义了三个配置类。每个类有单一职责：API详情、站点规则、发布目标。每个类都变成一个模块级单例实例。管道中的其他每个函数都从这些单例读取。 `` @dataclass class APIConfig: “”“HubEau数据获取的API配置。 可以把这看作是API的“通讯录”。 “”“ use_mock: bool = True base_url: str = “https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab” metric: str = “HIXnJ” # 每日最大水位（精细观测） # 分页：每次请求获取2万条记录（API限制） max_per_page: int = 20000 timeout_sec