Python 数据管线：自动化脚本也要有生产意识

Python 数据管线：自动化脚本也要有生产意识

一、脚本跑通不等于管线可靠

很多数据任务最初都是一个 Python 脚本：拉接口、清洗 CSV、写数据库、发报告。第一次跑通很有成就感，但真正麻烦在第二天、第三天和异常那天。接口超时怎么办，字段变了怎么办，重复执行会不会写脏数据，失败后能不能从中间恢复？

我经历过一个典型的"脚本级事故"：同事写了一个日报脚本，每天凌晨 2 点从第三方 API 拉订单数据，清洗后写入业务库。跑了三周都很稳定。某个周一早上，运营同学在群里 @ 我说"这周末的订单数据怎么全是上周五的？"原来周五那天脚本因为网络问题拉数据失败，没有报错，没有重试，文件是空的，但后续的清洗逻辑默默用前一天的数据覆盖了，而且因为报表模块没有检查数据日期，直接把错误数据推给了老板。一查才发现，源数据文件里只有 3 行标题和 0 条记录。

自动化脚本只要定时跑、影响业务，就应该按生产管线对待。日志、重试、幂等、告警和数据校验，一个都不能完全省。脚本跑通了只是开始，可靠地跑一年不出事故才是终点。

二、管线链路：每步都要可恢复

flowchart TD
    A[拉取数据] --> B[落原始区]
    B --> C[清洗转换]
    C --> D[质量校验]
    D --> E[写入目标表]
    E --> F[发送报告]
    D -->|校验失败| G[告警并阻塞下游]
    F --> H[记录运行元数据]

建议先落原始区，再清洗转换。直接从接口读完就写目标表，出了问题很难回放。原始数据保留一段时间，排查和重跑都会轻松很多。

原始区不只是"多存一份文件"。它承担三个角色：第一，它是回放源头，出了问题能从原始数据重新处理；第二，它是审计证据，下游数据异常时可以对比原始数据排查；第三，它是补数入口，支持按日期重新拉取历史数据。最好按日期分区存储，并记录每次拉取的元数据——时间、行数、源文件哈希、是否完整。

三、代码示例：写入前做简单校验

import pandas as pd
from datetime import date, timedelta
from typing import Optional

def validate_orders(df: pd.DataFrame, expect_date: Optional[date] = None) -> None:
    """订单数据校验，越早发现问题越好"""
    if df.empty:
        raise ValueError(f"数据为空，可能是源接口无返回")

    if df["order_id"].duplicated().any():
        dup_ids = df[df["order_id"].duplicated()]["order_id"].unique()
        raise ValueError(f"订单ID重复: {dup_ids[:5]}")

    if (df["amount"] < 0).any():
        neg_count = (df["amount"] < 0).sum()
        raise ValueError(f"{neg_count} 条订单金额为负")

    # 日期合理性检查
    if expect_date and df["order_date"].max() != expect_date:
        raise ValueError(
            f"期望数据日期 {expect_date}，实际最大日期 {df['order_date'].max()}"
        )

    # 数量级检查：数据量不该比日常少太多
    if len(df) < 10:  # 根据历史基线设置
        raise ValueError(f"数据量异常少: {len(df)} 条")

校验不一定复杂，但要覆盖关键业务约束。空数据、主键重复、金额异常、日期越界，这些问题越早发现越好。还有一个容易被忽略的：数据量级检查。如果日常订单是 5000 条，某天突然只有 10 条，很可能是接口分页出了问题，只拉了第一页。这种"部分数据"往往比"没有数据"更危险，因为下游会以为拿到了完整结果。

四、工程边界：幂等比重试更重要

管线失败后通常要重跑。如果任务不幂等，重跑可能重复写入、重复发消息、重复生成报告。可以按业务日期分区覆盖写入，也可以用唯一键 upsert，还可以把发送报告放在所有校验通过之后。重试前先保证重试安全。

取舍方面，简单脚本开发快，但状态管理弱；工作流平台更可靠，但有学习和维护成本。小团队可以先把脚本模块化：参数化日期、标准日志、错误退出码、结果校验。等任务数量多了，再迁到 Airflow、Dagster 或自研调度。

还要做告警分级。偶发接口超时可以自动重试，连续失败要通知人；数据校验失败要阻止下游报告；非核心报告延迟可以降级。所有失败都半夜叫醒人，最后没人看告警。告警要少而准。

数据管线还要保存运行元数据。每次任务处理了多少行、输入文件哈希是什么、输出分区是哪一天、质量校验结果如何，都应该记录。这样业务发现数字异常时，可以快速判断是源数据变了、清洗规则变了，还是任务没有跑完。

对外部接口要做契约防护。字段新增通常没事，字段删除、类型变化、枚举变化会直接影响清洗逻辑。可以在拉取后做 schema check，发现变化先报警，不要让错误数据进入下游。脚本越自动，越要防输入悄悄变。

报告发送要和数据写入解耦。数据没校验通过时，不要发送"成功报告"；报告发送失败时，也不要回滚已经正确写入的数据。动作边界清楚，重跑才安全。

管线还要支持手动补数。真实业务里经常会遇到某天接口失败、上游迟到、历史数据修正。补数入口要能指定日期范围，并且复用同一套校验和写入逻辑。不要单独写一份临时补数脚本，否则口径迟早分叉。

如果补数会影响已经发送的报告，还要标注版本并通知使用方。数据更新不是静悄悄的动作，它会影响决策。管线最后一步可以写入一个 sentinel 记录，标记"该日数据处理完成，数据版本 v2"，让下游报表系统能识别数据状态。

五、总结

Python 数据管线不要停留在"脚本能跑"。原始数据保留、质量校验、幂等重跑、分级告警和可恢复流程，才是自动化真正省心的关键。脚本写出来只花了半天，但它可能要跑三年。对三年负责，不是对第一天的 Demo 负责。

来源：https://blog.csdn.net/baronbool/article/details/162529132