从开发到生产

使用 Airflow 构建数据流水线

Volker Janz

Senior Developer Advocate at Astronomer

Airflow CLI

$ airflow dags trigger <dag_id>
$ airflow dags list-runs <dag_id>
$ airflow backfill create --dag-id <dag_id> ...

 

  • 与 UI 并用的生产工具包
  • 自动化运维并脚本化部署
使用 Airflow 构建数据流水线

触发一次运行

$ airflow dags trigger daily_sales_load --logical-date 2026-04-20

dag_id            | logical_date             | run_id
daily_sales_load  | 2026-04-20T00:00:00+00:00 | manual__2026-...

 

  • 为指定日期创建手动运行
  • 使用 --logical-date 指定运行要处理的日期
  • 注意 run_id 以 manual__ 开头
使用 Airflow 构建数据流水线

列出现有运行

$ airflow dags list-runs daily_sales_load

run_id                           | state   | logical_date
scheduled__2026-04-23T00:00:00   | success | 2026-04-23T00:00:00
manual__2026-04-20T00:00:00      | success | 2026-04-20T00:00:00

 

  • 显示该 Dag 的所有运行
  • run_id 前缀表明运行是如何创建的
  • scheduled__ = 由调度器创建
  • manual__ = 由trigger 命令或 UI 创建
使用 Airflow 构建数据流水线

回填历史数据

$ airflow backfill create \
    --dag-id daily_sales_load \
    --from-date 2026-04-20 \
    --to-date 2026-04-22 \
    --max-active-runs 1

 

  • 重新处理一段历史日期区间
  • 按计划区间逐区间创建一次运行
  • --max-active-runs 控制并发
  • 回填运行以 backfill__ 开头
使用 Airflow 构建数据流水线

三类运行

三种运行类型、前缀及其创建方式

  • scheduled__：由调度器自动创建
  • manual__：由trigger 命令或 UI 按钮创建
  • backfill__：由backfill 命令为历史日期创建
  • run_id 的前缀总能表明是哪一种
使用 Airflow 构建数据流水线

生产中的挑战

Airflow 生产挑战：构建、运行与观测

  • 构建：设置 Airflow 环境、AI 驱动的工作流开发、将代码部署到生产
  • 运行：扩展 worker，跨区域故障切换
  • 观测：故障排查、跟踪数据新鲜度、跨 Dag 及其外部的血缘追踪
使用 Airflow 构建数据流水线

Astro：构建

 

Astro CLI

  • 本地 Airflow 一条命令启动
  • 无缝部署到生产

$$

Astro IDE

  • 基于浏览器的 Dag 编写
  • AI 辅助编码，无需本地配置

Astro Build products

使用 Airflow 构建数据流水线

Astro：运行与观测

运行

  • 基于任务队列的弹性自动扩缩
  • 高可用与自动故障切换
  • 无需自管基础设施

观测

  • 跨 Dag 与表的管道血缘
  • 主动 SLA 告警，提前预警逾期
  • AI 日志摘要，更快定位根因

Astro Observe

使用 Airflow 构建数据流水线

让我们一起练习吧！

使用 Airflow 构建数据流水线

Preparing Video For Download...