data-pipelines

数据管道技能

版本： 1.0 技术栈： Airflow、Step Functions、dbt、通用 ETL

管道可能会悄无声息地失败。非幂等任务每次重试都会追加重复行。未经验证的任务会将错误数据传递到下游，直到三阶段之后仪表板显示出不可能的数字，你才会发现问题。没有参数化日期的管道无法回填，这意味着周二出了问题时，你必须手动重新处理自上次确认无误以来的每一天。

幂等任务、阶段间质量检查和参数化执行意味着故障可以恢复，错误能在发生时被发现，而不是等到下游才暴露。

---

范围与边界

此技能涵盖：

• ETL/ELT 架构决策
• 编排模式（DAG、依赖关系）
• 幂等性与重试策略
• 数据质量检查
• 增量刷新与全量刷新
• 回填模式

交由其他技能处理：

• data-python：任务中的 Python 代码
• data-sql：转换中的查询模式
• data-aws：AWS 专属服务模式

适用场景： 设计或构建数据管道。

---

核心原则

1. 幂等任务 — 运行两次会产生相同结果。
2. 原子操作 — 要么全部完成，要么全部不完成；不产生部分写入。
3. 数据契约 — 定义各阶段之间的架构预期。
4. 可观测的管道 — 在每个阶段记录日志、指标并设置告警。
5. 尽可能采用增量处理 — 只处理发生变化的数据。

---

模式

幂等写入

# Bad - appends duplicate data on retry
def load_data(df, table):
    df.to_sql(table, engine, if_exists="append")

# Good - replace partition, safe to retry
def load_data(df, table, partition_date):
    with engine.begin() as conn:
        conn.execute(f"DELETE FROM {table} WHERE date = '{partition_date}'")
        df.to_sql(table, conn, if_exists="append", index=False)

阶段模式

raw/           # Exact copy of source, immutable
├── source_a/
└── source_b/

staged/        # Cleaned, typed, validated
├── source_a/
└── source_b/

curated/       # Business logic applied, joined
├── dim_users/
├── dim_products/
└── fact_orders/

aggregated/    # Pre-computed metrics
├── daily_revenue/
└── user_cohorts/

数据质量检查

def validate_output(df: pd.DataFrame) -> None:
    """Run after transform, before load."""
    assert not df.empty, "Output is empty"
    assert df["id"].is_unique, "Duplicate IDs found"
    assert df["amount"].notna().all(), "NULL amounts found"
    assert (df["amount"] >= 0).all(), "Negative amounts found"

---

反模式

反模式	问题	修复方法
非幂等任务	重试时产生重复数据	先 DELETE 再 INSERT，或使用 upsert
单体 DAG	难以调试，恢复时间长	拆分为更小、更专注的 DAG
不验证数据	错误数据悄然传播	在各阶段之间添加检查
日期硬编码	无法回填	对执行日期进行参数化
静默失败	问题发现得太晚	对异常情况而不只是错误发出告警

---

检查清单

• 所有任务均为幂等任务
• 在各阶段之间进行数据质量检查
• 执行日期已参数化
• 已测试并记录回填流程
• 对故障和异常情况均设置告警
• 已记录清晰的数据血缘

---

参考资料

• references/idempotency.md — 幂等写入模式与测试

资源

• assets/pipeline-checklist.md — 全面的管道设计检查清单