Python 数据管线事故复盘：为何一个脚本错误影响了全链路

一、周五下午 4:50 部署的数据脚本，周六凌晨整个数据仓库崩了

事故经过：

• 周五 16:50 数据分析师提交了一个新增的"用户行为标签计算"脚本
• 周六 02:00 例行 ETL 任务启动，新脚本作为 DAG 的一个节点投入运行
• 周六 02:45 Airflow 报错：Task 失败，下游 12 个 Task 全部阻塞
• 周六 06:30 值班人员被告警叫醒，开始排查
• 周六 08:00 定位到问题：新脚本在空数据集上执行了 pandas 除法操作
• 周六 09:00 回滚脚本，手动补跑昨日数据

根因是什么？不是代码写得烂，而是数据管线的"链式依赖"设计没有考虑单节点失败的隔离性。

二、事故的根因分析

链条很清晰：一个 pandas 除零错误 → Task 失败 → 下游全挂。但真正的问题是：为什么一个非核心字段的计算错误会阻塞核心的 BI 报表？答案是 DAG 依赖设计把"强依赖"和"弱依赖"混在了一起。

三、错误代码与修复

# ❌ 事故代码（数据工程师原版）
def calculate_user_activity(user_df):
    """
    计算用户活跃度分数
    事故点: 注册天数为0时，除法产生异常
    """
    user_df['activity_score'] = (
        user_df['login_days'] / user_df['registered_days']
    )
    user_df['activity_tier'] = pd.cut(
        user_df['activity_score'],
        bins=[0, 0.2, 0.5, 0.8, float('inf')],
        labels=['low', 'medium', 'high', 'power']
    )
    return user_df

# ✅ 修复后的代码
def calculate_user_activity_robust(user_df):
    """
    计算用户活跃度分数（数据安全版本）
    """
    df = user_df.copy()
    
    # 1. 输入校验
    required_cols = ['login_days', 'registered_days']
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        raise ValueError(f"缺少必要列: {missing}")
    
    # 2. 异常记录日志
    zero_mask = df['registered_days'] == 0
    if zero_mask.any():
        logging.warning(
            f"发现 {zero_mask.sum()} 条记录注册天数为0, "
            f"user_ids={df.loc[zero_mask, 'user_id'].tolist()[:10]}"
        )
    
    # 3. 安全计算（除零保护）
    df['activity_score'] = df.apply(
        lambda row: (
            row['login_days'] / row['registered_days']
            if row['registered_days'] > 0 else None  # 无数据标记为 None
        ),
        axis=1
    )
    
    # 4. 分箱操作的空值保护
    valid_mask = df['activity_score'].notna()
    if valid_mask.sum() == 0:
        logging.warning("所有记录的活跃度分数都无法计算")
        df['activity_tier'] = 'unknown'
        return df
    
    df.loc[valid_mask, 'activity_tier'] = pd.cut(
        df.loc[valid_mask, 'activity_score'],
        bins=[0, 0.2, 0.5, 0.8, float('inf')],
        labels=['low', 'medium', 'high', 'power']
    ).astype(str)
    
    df['activity_tier'] = df['activity_tier'].fillna('unknown')
    
    # 5. 输出数据质量报告
    stats = {
        'total': len(df),
        'valid': valid_mask.sum(),
        'null_rate': (~valid_mask).mean(),
        'tier_distribution': df['activity_tier'].value_counts().to_dict(),
    }
    logging.info(f"活跃度计算完成: {stats}")
    
    return df

# ✅ DAG 依赖的修复
# 之前: Task A >> Task B >> Task C (全串联)
# 之后: 弱依赖用 trigger_rule
"""
task_a = calculate_activity()
task_bi = generate_bi_report()
task_recommend = update_recommend_features()

# 关键修改: BI 报表不因 activity 计算失败而阻塞
task_a >> task_recommend  # 推荐依赖 activity（强依赖）
task_bi  # BI 报表独立运行（无依赖）
# 或使用 Airflow 的 trigger_rule
task_recommend.trigger_rule = 'one_failed'  # 即使上游失败也继续
"""

四、系统性改进措施

数据管线的"熔断"设计：每个 Task 应该有独立的异常处理，不应该把 pandas 的原生异常直接暴露给 Airflow。所有数据操作都应该包装在 try-except 中，将异常转化为可观测的指标（如 null_rate 增加），而不是 Task 失败。

依赖分级：强依赖（下游必须等上游完成才能跑）用 >> 串行，弱依赖（上游失败了也能带着不完整数据跑）用 trigger_rule='one_failed'。这样即使行为标签没算出来，核心的营收报表仍然能准时生成。

数据质量前置检查：在 Task 执行前加一个"数据网关"——快速检查输入数据的基本质量（非空率、数据量波动、关键列是否存在）。质量不达标时，发送告警并暂停执行，而不是等跑到一半才发现数据有问题。

部署和回滚流程：数据管线的代码变更应该有"金丝雀"发布——先在测试环境跑一次全量数据，然后才上线。回滚方面，保留最近 3 个版本的脚本代码，回滚操作不需要重新部署——只需要在 Airflow 中切换 Task 的脚本路径。

五、总结

这次事故根因是"数据操作缺乏防御性编程"和"DAG 依赖缺乏容错"。修复分三层：代码层（所有数学运算加除零保护、空值检查）、管线层（强依赖和弱依赖分级）、流程层（上线前必须跑全量数据测试）。最关键的认知：数据管线不是"Script 的集合"，而是"数据产品的生产线"。生产线上的任何一个环节都要有"部分降级"的能力——断了一条辅线，主线还得跑。