ETL/ELT 编排与批处理：从 DAG 到可重跑

0. 元信息

• 主题路径：docs/topics/data-engineering-basics/subtopics/etl-and-batch-pipelines/README.md
• 父主题：data-engineering-basics
• 适合对象：会写 Python/SQL、跑过 cron 脚本、希望从”散装脚本”过渡到”可治理的批处理流水线”的工程师
• 建议周期：1.5～2 周，每周 10～14 小时
• 前置知识：Python、SQL、Docker；建议先完成父主题第 1 阶段
• 最终目标：能为给定业务需求设计一条可重跑、可回填、可观测的 ETL/ELT DAG，并在 Airflow/Prefect/Dagster 三个编排器中至少精通一个

1. 学习路线

DAG 与 Task 模型 → Operator 与 Hook → 调度与回填 → 幂等与状态管理 → 可观测与告警 → 跨编排器对比

2. 阶段周数分配

每天 1.5～2 小时。1.5 周方案专注前 5 阶段；2 周方案多一天做跨编排器对比与故障演练。

3. 核心知识 / 产出 / 标准表

4. 第一周（每天 1.5~2 小时）

Day 1 约定：本计划以 Linux + Docker + 8G 内存为最小实验环境。所有流水线在 Docker Compose 内启动，代码 SQL 配置进 Git。所有任务必须可重跑：清空状态后再跑产出相同结果。

Day 7 执行次序

步骤 A —— 最小可用流水线（60~90 分钟）

1. CSV → PG ODS（PythonOperator + to_sql）；
2. dbt 写 stg_orders.sql（清洗）、dwd_orders.sql（明细）、dws_orders_daily.sql（每日汇总）；
3. 三层 dbt 全部 dbt build 通过；
4. 验证：行数从 ODS 到 ADS 逐层变化合理（dbt show 或 psql）。

步骤 B —— 4 类边界用例（30~45 分钟）

Day 7 当天必完成步骤 A；步骤 B 至少完成 B1、B3。

5. 阶段通用验收

1. 不看答案独立重写一条 5 任务 ETL DAG + 一次 backfill 演练；
2. 用自己的话解释”为什么 DAG 必须是 DAG""为什么需要 idempotency""为什么 backfill 与 catchup 是不同操作”；
3. 画一张图：DAG 拓扑序图 + 重试状态机 + 数据流图（三件套之一）；
4. 测试空 CSV、列缺失、Schema 漂移、分区重复、Task 失败 5 类边界；
5. 准备至少 3 组自定义 CSV 数据并贴出实际行数与耗时；
6. 记录 DAG 端到端 P50 / P99 耗时、Task 失败率、backfill 时长、告警响应时间 4 个核心指标；
7. 能修改已有 DAG（加一层、加一个 Task、换一种 Operator）并用同样指标验证。

交付存放：第 3 项的图、第 5 项的输出、第 6 项的指标，统一存到 week1/notes/ 或 week1/<dag>/README.md。

6. 最终验收

• 独立设计并交付一条可重跑、可回填、可观测的 ETL/ELT DAG，覆盖至少 3 个分层、5 类数据测试、2 种告警通道；

• 至少完成 18 个实战项（分布建议）：

  子阶段	题目数量	难度	平台建议
  DAG 模型与 Operator	3	Easy	Astronomer Registry
  调度与回填	3	Easy / Medium	Airflow Tutorial
  幂等与状态管理	4	Medium	Astronomer Registry + 自己改写
  可观测与告警	4	Medium	StatsD + Prometheus + Slack
  跨编排器对比	4	Medium / Hard	Dagster + Prefect + Airflow 互译

  约束：至少 9 项达到 Medium，至少 3 项达到 Hard；每项必须留可运行的代码与说明文档。

• 完成 1 个综合项目：mini 数仓 ETL（CSV → Iceberg ODS → dbt DWD/DWS/ADS），含分区 + 幂等 + 测试 + 告警；

• 能用 15 分钟讲清 DAG 拓扑序、Operator 与 Hook 边界、idempotency 实现、backfill 与 catchup 区别、SLI/SLO 设定。

7. 综合项目

首选：mini 数仓 + ETL + dashboard（必做：分层数仓 + dbt 转换 + Streamlit 看板）。
备选：CDC + 实时 dashboard（Debezium → Kafka → Flink → Postgres → Superset）。
备选：把遗留 cron 脚本改造成 Airflow DAG（含幂等、测试、告警）。

mini 数仓必做要求：

• 输入：订单 CSV（用户 / 商品 / 金额 / 时间），按天分区，至少 3 份不同 schema 的 CSV；
• 输出：必输出（1）三层 schema：ODS（原始）→ DWD（清洗 + SCD Type 2）→ DWS（每日汇总）→ ADS（Top 商品 / 用户 ARPU）；（2）Streamlit 看板：日订单量 / GMV / Top 10 商品；（3）dbt tests 5 条（unique / not_null / relationships / freshness / accepted_values）；（4）Airflow 告警通道（Slack 或邮件）；
• 算法 / 工程：DAG 至少 7 Task（extract → validate → load_ods → dbt_run → dbt_test → load_ads → notify）；用 EXPLAIN ANALYZE 看每条关键 SQL 计划；
• 进阶可选：接 Great Expectations 加 5 条分布异常检测；接 OpenLineage 暴露血缘事件。

任何综合项目都必须包含：

1. 需求说明与数据契约（源 schema、目标 schema、SLA）；
2. DAG 架构图与命名规范；
3. 核心代码（Airflow DAG + dbt models + 告警 callback）；
4. 边界测试（源缺失、Schema 漂移、分区重复、Task 失败、dbt 测试失败）；
5. 可观测（StatsD 指标 + Prometheus 抓取 + Grafana 看板）；
6. README（设计取舍、已知限制、下一步）；
7. 复盘记录（模拟 1 次失败，写 retrospective.md）；
8. notes/ 规范：

本主题贡献

ETL/ELT 编排的核心矛盾是”批处理天然不幂等 vs 业务要求可重跑可回填”。本子主题专门讲清 Airflow DAG 如何把这条矛盾变成可治理流程——以 Airflow DAG 为编排骨架，用 partition-overwrite 把 Task 变幂等，用 dbt + Spark 做转换，把 SLA 与告警接到 StatsD/Prometheus；不重复父主题的 4 层分层与通用 dbt 介绍。

3 职责

1. 用 Airflow DAG TaskFlow API 编排 extract → validate → load_ods → dbt_run → dbt_test → load_ads → notify 7 步，把 cron 脚本搬进有向无环图，每个 Task 配 retries=3 / retry_delay=timedelta(minutes=5) / sla=timedelta(minutes=30)。
2. 用 partition-overwrite + 幂等键（按 logical_date 分区 delete-then-insert）保证同一日期多次触发产出相同；下游 dbt incremental 与 Spark 批读按日期分区对齐，Task 间不传大数据走 XCom 写 S3。
3. 用 StatsD + Prometheus 暴露 DAG duration / task failure rate / SLA miss 三类指标，配 on_failure_callback 发 Slack 告警，把批处理流水线接进 on-call 体系。

4 交付物

1. 一条 7 任务 Airflow DAG（extract_csv → validate_schema → load_ods → dbt_run → dbt_test → load_ads → notify），TaskFlow API + PostgresOperator + PythonOperator 混用，配 schedule="0 2 * * *" 与 catchup=False。
2. 一份 partition-overwrite 幂等脚本（DELETE FROM orders WHERE order_date = {{ ds }} 后再 INSERT），含 airflow dags backfill --start-date --end-date 演练命令与”重跑 2 次不变行数”验证。
3. 一份 dbt 项目（staging / intermediate / marts 三层），5 条 dbt test（unique / not_null / accepted_values / relationships / freshness），dbt build exit code 0 作为下游 Task 的成功门槛。
4. 一份 SLA 文档 + Slack 告警 callback（含 PagerDuty 路由），4 个核心指标基线：DAG P50/P99 耗时 / Task 失败率 / backfill 时长 / 告警响应时间。

3 指标

1. 重跑幂等率 100%（同一 logical_date 触发 2 次，COUNT(*) 与聚合值完全一致）。
2. DAG 端到端 P99 耗时 < 30 min（按数据量级配 executor + pool + priority_weight）。
3. SLA miss 告警到达 Slack < 1 min（on_failure_callback 链路耗时，含 webhook 重试）。

8. 推荐开源资料

许可证提示：Astronomer Registry 中 DAG 模板以 Apache-2.0 为主；复制或大幅参考前请打开 LICENSE 确认。Airflow、Dagster、Prefect 自身都是 Apache-2.0；商用无需特殊授权。默认做法是读思路后自己重写，而不是复制粘贴。

默认使用顺序：先读 Beauchemin《Functional Data Engineering》建立”不可变数据”心智模型 → 装 Airflow 跑通第一条 DAG → 读 Harenslak 第 1~6 章 → 跑 Astronomer Registry 模板 3 条并改成自己的业务 → 加 dbt 做转换与测试 → 加 StatsD + Prometheus 暴露指标 → 用 crontab.guru 验证 cron → 对比 Dagster / Prefect 的 asset 模型 → 写复盘到 notes/retrospective.md。

9. 学习资料汇聚（v0.3 自包含）

本节由本计划生成。链接指向原始材料或作者公开内容。规范会演进，记录时务必写明版本与日期。

9.1 背景与动机

数据从产生到可用，中间涉及定时拉取、清洗、转换、写入多个步骤。手工 cron 脚本散落各处，重跑靠运气，失败靠人盯。Airflow 2014 年由 Airbnb 开源，把”有向无环图（DAG）“作为批处理核心抽象，引入了依赖、调度、回填、可观测四大支柱。Prefect 2018 年、Dagster 2019 年先后开源，各自提出”动态工作流""Software-defined Assets”等新抽象。今天数据团队在五个编排器中做选择时，必须回答”我要 task-based 还是 asset-based”。

9.2 概念地图

flowchart LR
  Cron[cron 脚本] --> DAG[DAG]
  DAG --> Task[Task]
  Task --> Op[Operator]
  Task --> Hook[Hook]
  DAG --> Schedule[Schedule cron/catchup]
  DAG --> Backfill[Backfill]
  Task --> XCom[XCom]
  Task --> Idem[幂等键]
  DAG --> SLA[SLA miss]
  DAG --> Alert[告警]
  DAG --> Asset[Asset asset-based]

9.3 基础知识讲解

9.3.1 经典论文与规范

9.3.2 经典书籍

9.3.3 优秀博客与文档

9.3.4 核心人物

9.3.5 开发方法

9.4 经典问题与经典案例（≥5 道）

9.5 学习难点

概念难点

思维难点

工程难点

9.6 技术标准与接口

Entity

Scope

Airflow/Prefect/Dagster 是编排器，不替代存储与计算；Argo 偏 Kubernetes-native；Temporal 偏微服务编排。

Structure

• Airflow：DAG → Task → Operator/Hook → Sensor，Task 间用 XCom 传值。
• Dagster：Asset → Op → Job，Asset 是数据产物。
• Prefect：Flow → Task，Flow 是 Python 函数。
• Argo：WorkflowTemplate → Workflow → DAG → Task。

Ecosystem

• DAG 模板：Astronomer Registry、Airflow Providers。
• 告警：Slack、PagerDuty、Email。
• 监控：Prometheus、StatsD、OpenLineage。

Depth Tiers

本子主题目标：L3。

Source

• Apache Airflow 官方文档：引用快照 2026-07-30。
• Dagster 官方文档：引用快照 2026-07-30。
• Astronomer Registry：引用快照 2026-07-30。

9.7 关键代码

9.7.1 Airflow TaskFlow API DAG（5 任务 ETL）

# 9.7.1 Airflow TaskFlow API 写一条 CSV -> Parquet -> PostgreSQL 的 DAG
from datetime import datetime, timedelta
from airflow.decorators import dag, task
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

@dag(
    dag_id="etl_orders_daily",
    start_date=datetime(2026, 1, 1),
    schedule="0 2 * * *",        # 每日 02:00
    catchup=False,
    retries=3,
    retry_=timedelta(minutes=5),
    default_args={"owner": "data-eng"},
)
def etl_orders():
    @task
    def extract() -> str:
        df = pd.read_csv("/data/orders.csv")
        path = f"/data/orders_{datetime.utcnow():%Y%m%d}.parquet"
        pq.write_table(pa.Table.from_pandas(df), path)
        return path

    @task
    def transform(parquet_path: str) -> pd.DataFrame:
        df = pq.read_table(parquet_path).to_pandas()
        df["total"] = df["qty"] * df["price"]
        return df

    @task
    def load(df: pd.DataFrame) -> int:
        # 幂等：先删当月分区再写
        from sqlalchemy import create_engine
        engine = create_engine("postgresql://airflow:airflow@postgres:5432/dw")
        with engine.begin() as conn:
            conn.exec_driver_sql(
                f"DELETE FROM orders WHERE order_date = '{df['order_date'].iloc[0]}'"
            )
        df.to_sql("orders", engine, if_exists="append", index=False)
        return len(df)

    @task
    def notify(row_count: int) -> None:
        print(f"loaded {row_count} rows")

    p = extract()
    t = transform(p)
    n = load(t)
    notify(n)

etl_orders()

9.7.2 Dagster Software-defined Asset

# 9.7.2 Dagster asset-based ETL（同一份代码，不同哲学）
from dagster import asset, Definitions
import pandas as pd

@asset(group_name="etl")
def orders_raw() -> pd.DataFrame:
    return pd.read_csv("/data/orders.csv")

@asset(group_name="etl")
def orders_clean(orders_raw: pd.DataFrame) -> pd.DataFrame:
    df = orders_raw.copy()
    df["total"] = df["qty"] * df["price"]
    return df

@asset(group_name="etl")
def orders_summary(orders_clean: pd.DataFrame) -> pd.DataFrame:
    return orders_clean.groupby("region")["total"].sum().reset_index()

defs = Definitions(assets=[orders_raw, orders_clean, orders_summary])

9.7.3 Backfill 与幂等性验证脚本

# 9.7.3 用 Airflow CLI 触发 backfill 并验证幂等性
airflow dags backfill etl_orders_daily \
    --start-date 2026-01-01 --end-date 2026-01-07

# 验证：同一日期再跑一次，对比 row_count 不变
airflow dags trigger etl_orders_daily -e 2026-01-01
psql -h postgres -U airflow -d dw -c \
  "SELECT order_date, COUNT(*) FROM orders GROUP BY 1 ORDER BY 1;"

10. 常见误区

• 把 cron 脚本直接搬进 Airflow，不做幂等；
• Task 用全局变量传值，不走 XCom；
• Backfill 与 catchup 混为一谈；
• Sensor 写死轮询，不配 mode="reschedule"；
• SLA 设了不告警；
• Task 失败只重试 1 次就放弃；
• DAG 文件大而全，缺 TaskGroup；
• 用 BashOperator 调用 Python 脚本（应直接 PythonOperator/@task）；
• 不挂 OpenLineage，缺血缘；
• 跨编排器随意切换，团队没人精通；
• 误把 data_interval_start/end 当 execution_date；
• Backfill 用 catchup=True 而非显式命令。

11. 所有知识点分类

1. 编程语言
2. 数据结构与算法
3. 计算机基础
4. 工程技术
5. Web 与后端
6. 前端与客户端
7. 数据与人工智能
8. 项目与职业能力
9. 安全与可靠性

本计划归属：数据与人工智能 主 + 工程技术 辅。