airflow

Airflow 运维

使用 af 命令查询、管理和排查 Airflow 工作流。

Astro CLI

Astro CLI 是在本地运行 Airflow 并部署到生产环境的推荐方式。它提供开箱即用的容器化 Airflow 环境：

# Initialize a new project
astro dev init

# Start local Airflow (webserver at http://localhost:8080)
astro dev start

# Parse DAGs to catch errors quickly (no need to start Airflow)
astro dev parse

# Run pytest against your DAGs
astro dev pytest

# Deploy to production
astro deploy            # Full deploy (image + DAGs)
astro deploy --dags     # DAG-only deploy (fast, no image build)

更多详情：

• 新建项目？ 请参阅 setting-up-astro-project 技能
• 配置本地环境？ 请参阅 managing-astro-local-env 技能
• 部署？ 请参阅 deploying-airflow 技能

---

运行 CLI

以下命令假定 af 已添加到 PATH。运行 astro otto 即可自动获取，或使用 uv tool install astro-airflow-mcp 安装独立版本。

实例配置

通过持久化配置管理多个 Airflow 实例：

# Add a new instance
af instance add prod --url https://airflow.example.com --token "$API_TOKEN"
af instance add staging --url https://staging.example.com --username admin --password admin

# List and switch instances
af instance list      # Shows all instances in a table
af instance use prod  # Switch to prod instance
af instance current   # Show current instance
af instance delete old-instance

# Auto-discover instances (use --dry-run to preview first)
af instance discover --dry-run        # Preview all discoverable instances
af instance discover                  # Discover from all backends (astro, local)
af instance discover astro            # Discover Astro deployments only
af instance discover astro --all-workspaces  # Include all accessible workspaces
af instance discover local            # Scan common local Airflow ports
af instance discover local --scan     # Deep scan all ports 1024-65535

# IMPORTANT: Always run with --dry-run first and ask for user consent before
# running discover without it. The non-dry-run mode creates API tokens in
# Astro Cloud, which is a sensitive action that requires explicit approval.

# Show where an instance came from (file path + scope)
af instance show prod

# Override instance for a single command via env vars
AIRFLOW_API_URL=https://staging.example.com AIRFLOW_AUTH_TOKEN=$STG af dags list

# Or switch persistently
af instance use staging

配置布局（与 git config 的 system/global/local 布局一致）：

作用域	文件	是否提交？
全局	~/.astro/config.yaml	不适用（每个用户独有）
项目共享	<root>/.astro/config.yaml	是
项目本地	<root>/.astro/config.local.yaml	否（已添加到 gitignore）

通过从 cwd 向上查找 .astro/ 来确定 <root>。在项目内默认的写入路由为：add/discover → 项目共享，use → 项目本地。使用 --global / --project / --local 覆盖。设置 AF_CONFIG=<path> 可绕过分层配置并使用单个文件。

使用 af migrate 从旧版 ~/.af/config.yaml 迁移（可重复执行；会将旧文件重命名为 .bak）。

配置中的令牌可以使用 ${VAR} 语法引用环境变量：

instances:
- name: prod
  url: https://airflow.example.com
  auth:
    token: ${AIRFLOW_API_TOKEN}

也可以直接使用环境变量（无需配置文件）：

export AIRFLOW_API_URL=http://localhost:8080
export AIRFLOW_AUTH_TOKEN=your-token-here
# Or username/password:
export AIRFLOW_USERNAME=admin
export AIRFLOW_PASSWORD=admin

或者使用 CLI 标志：af --airflow-url http://localhost:8080 --token "$TOKEN" <command>

快速参考

命令	说明
af health	检查系统健康状况
af dags list	列出所有 DAG
af dags get <dag_id>	获取 DAG 详情
af dags explore <dag_id>	全面排查 DAG
af dags source <dag_id>	获取 DAG 源代码
af dags pause <dag_id>	暂停 DAG 调度
af dags unpause <dag_id>	恢复 DAG 调度
af dags errors	列出导入错误
af dags warnings	列出 DAG 警告
af dags stats	DAG 运行统计信息
af runs list	列出 DAG 运行记录
af runs get <dag_id> <run_id>	获取运行详情
af runs trigger <dag_id>	触发 DAG 运行
af runs trigger-wait <dag_id>	触发 DAG 运行并等待完成
af runs delete <dag_id> <run_id>	永久删除 DAG 运行记录
af runs clear <dag_id> <run_id>	清除运行记录以便重新执行
af runs diagnose <dag_id> <run_id>	诊断失败的运行
af tasks list <dag_id>	列出 DAG 中的任务
af tasks get <dag_id> <task_id>	获取任务定义
af tasks instance <dag_id> <run_id> <task_id>	获取任务实例
af tasks logs <dag_id> <run_id> <task_id>	获取任务日志
af config version	Airflow 版本
af config show	完整配置
af config connections	列出连接
af config variables	列出变量
af config variable <key>	获取指定变量
af config pools	列出池
af config pool <name>	获取池详情
af config plugins	列出插件
af config providers	列出提供程序
af config assets	列出资产/数据集
af api <endpoint>	直接访问 REST API
af api ls	列出可用的 API 端点
af api ls --filter X	列出与模式匹配的端点
af registry providers	列出 Airflow Registry 中的提供程序
af registry modules <provider>	列出提供程序中的运算符/钩子/传感器/传输器
af registry parameters <provider>	提供程序类的构造函数签名（名称、类型、默认值、是否必需）
af registry connections <provider>	列出提供程序公开的连接类型

用户意图模式

入门

• “如何在本地运行 Airflow？” / “设置 Airflow” -> 使用 managing-astro-local-env 技能（使用 Astro CLI）
• “创建新的 Airflow 项目” / “初始化项目” -> 使用 setting-up-astro-project 技能（使用 Astro CLI）
• “如何安装 Airflow？” / “开始使用 Airflow” -> 使用 setting-up-astro-project 技能

DAG 操作

• “有哪些 DAG？” / “列出所有 DAG” -> af dags list
• “介绍一下 DAG X” / “DAG Y 是什么？” -> af dags explore <dag_id>
• “DAG X 的计划是什么？” -> af dags get <dag_id>
• “显示 DAG X 的代码” -> af dags source <dag_id>
• “停止 DAG X” / “暂停此工作流” -> af dags pause <dag_id>
• “恢复 DAG X” -> af dags unpause <dag_id>
• “是否有 DAG 错误？” -> af dags errors
• “创建新的 DAG” / “编写管道” -> 使用 authoring-dags 技能

运行操作

• “执行过哪些运行？” -> af runs list
• “运行 DAG X” / “触发管道” -> af runs trigger <dag_id>
• “运行 DAG X 并等待” -> af runs trigger-wait <dag_id>
• “这次运行为什么失败？” -> af runs diagnose <dag_id> <run_id>
• “删除这次运行” / “移除卡住的运行” -> af runs delete <dag_id> <run_id>
• “清除这次运行” / “重试这次运行” / “重新运行” -> af runs clear <dag_id> <run_id>
• “测试此 DAG，若失败则修复” -> 使用 testing-dags 技能

任务操作

• “DAG X 中有哪些任务？” -> af tasks list <dag_id>
• “获取任务日志” / “任务为什么失败？” -> af tasks logs <dag_id> <run_id> <task_id>
• “全面分析根本原因” / “诊断并修复” -> 使用 debugging-dags 技能

数据操作

• “数据是最新的吗？” / “此表上次更新时间是什么时候？” -> 使用 checking-freshness 技能
• “这些数据来自哪里？” -> 使用 tracing-upstream-lineage 技能
• “哪些内容依赖此表？” / “如果我更改此表，会有什么受影响？” -> 使用 tracing-downstream-lineage 技能

部署操作

• “部署我的 DAG” / “推送到生产环境” -> 使用 deploying-airflow 技能
• “设置 CI/CD” / “自动化部署” -> 使用 deploying-airflow 技能
• “部署到 Kubernetes” / “设置 Helm” -> 使用 deploying-airflow 技能
• “astro deploy” / “仅部署 DAG” -> 使用 deploying-airflow 技能

系统操作

• “Airflow 是什么版本？” -> af config version
• “有哪些连接？” -> af config connections
• “资源池是否已满？” -> af config pools
• “Airflow 是否运行正常？” -> af health

API 探索

• “有哪些可用的 API 端点？” -> af api ls
• “查找变量端点” -> af api ls --filter variable
• “访问 XCom 值” / “获取 XCom” -> af api xcom-entries -F dag_id=X -F task_id=Y
• “获取事件日志” / “审计跟踪” -> af api event-logs -F dag_id=X
• “通过 API 创建连接” -> af api connections -X POST --body '{...}'
• “通过 API 创建变量” -> af api variables -X POST -F key=name -f value=val

注册表发现

• “提供程序 X 有哪些运算符？” -> af registry modules <provider>
• “运算符 Y 的构造函数参数有哪些？” -> af registry parameters <provider>
• “有哪些提供程序？” / “是否有适用于 Z 的提供程序？” -> af registry providers
• “提供程序 X 提供哪些连接类型？” -> af registry connections <provider>
• “使用特定运算符编写 DAG” -> 复制示例前，使用注册表核实当前签名

常见工作流

部署前验证 DAG

如果你使用 Astro CLI，无需运行中的 Airflow 实例即可验证 DAG：

# Parse DAGs to catch import errors and syntax issues
astro dev parse

# Run unit tests
astro dev pytest

否则，请针对运行中的实例进行验证：

af dags errors     # Check for parse/import errors
af dags warnings   # Check for deprecation warnings

编写代码前先查看 Operator 签名

Airflow Registry（位于 airflow.apache.org/registry）是提供程序类及其当前构造函数签名的权威来源。编写 DAG 时，请优先参考它，而不是依赖记忆或过时的文档——Registry 反映的是当前发布的提供程序版本。

# List all providers and pick the one you need
af registry providers | jq '.providers[] | {id, name, version}'

# List every operator / hook / sensor in a provider (e.g. standard, amazon, google)
af registry modules standard \
  | jq '.modules[] | {name, type, import_path, docs_url}'

# Get the current constructor signature for a specific class
af registry parameters standard \
  | jq '.classes["airflow.providers.standard.operators.hitl.ApprovalOperator"].parameters'

# Filter modules by substring (useful when you know the concept but not the class)
af registry modules standard \
  | jq '.modules[] | select(.import_path | test("hitl"))'

结果会缓存在本地：最新版本缓存 1 小时，固定版本缓存 30 天（固定版本不可变）。在任何 modules / parameters / connections 调用中添加 --version X.Y.Z，即可指定目标版本。

调查运行失败的原因

# 1. List recent runs to find failure
af runs list --dag-id my_dag

# 2. Diagnose the specific run
af runs diagnose my_dag manual__2024-01-15T10:00:00+00:00

# 3. Get logs for failed task (from diagnose output)
af tasks logs my_dag manual__2024-01-15T10:00:00+00:00 extract_data

# 4. After fixing, clear the run to retry all tasks
af runs clear my_dag manual__2024-01-15T10:00:00+00:00

每日晨间健康检查

# 1. Overall system health
af health

# 2. Check for broken DAGs
af dags errors

# 3. Check pool utilization
af config pools

了解 DAG

# Get comprehensive overview (metadata + tasks + source)
af dags explore my_dag

检查 DAG 未运行的原因

# Check if paused
af dags get my_dag

# Check for import errors
af dags errors

# Check recent runs
af runs list --dag-id my_dag

触发并监控

# Option 1: Trigger and wait (blocking)
af runs trigger-wait my_dag --timeout 1800

# Option 2: Trigger and check later
af runs trigger my_dag
af runs get my_dag <run_id>

输出格式

所有命令均输出 JSON（instance 命令除外，它们使用易读的表格）：

af dags list
# {
#   "total_dags": 5,
#   "returned_count": 5,
#   "dags": [...]
# }

使用 jq 进行筛选：

# Find failed runs
af runs list | jq '.dag_runs[] | select(.state == "failed")'

# Get DAG IDs only
af dags list | jq '.dags[].dag_id'

# Find paused DAGs
af dags list | jq '[.dags[] | select(.is_paused == true)]'

任务日志选项

# Get logs for specific retry attempt
af tasks logs my_dag run_id task_id --try 2

# Get logs for mapped task index
af tasks logs my_dag run_id task_id --map-index 5

使用 af api 直接访问 API

对于高级命令未覆盖的端点（XCom、event-logs、backfills 等），请使用 af api。

# Discover available endpoints
af api ls
af api ls --filter variable

# Basic usage
af api dags
af api dags -F limit=10 -F only_active=true
af api variables -X POST -F key=my_var -f value="my value"
af api variables/old_var -X DELETE

字段语法：-F key=value 会自动转换类型，-f key=value 则保留为字符串。

完整参考：请参阅 api-reference.md，了解所有选项、常见端点（XCom、event-logs、backfills）和示例。

相关技能

技能	适用场景……
authoring-dags	遵循最佳实践创建或编辑 DAG 文件
testing-dags	反复执行测试 -> 调试 -> 修复 -> 重新测试
debugging-dags	深入分析根本原因并诊断故障
checking-freshness	检查数据是否为最新或已过时
tracing-upstream-lineage	查找数据的来源
tracing-downstream-lineage	影响分析——更改某项内容会导致什么失效
deploying-airflow	将 DAG 部署到生产环境（Astro、Docker Compose、Kubernetes）
migrating-airflow-2-to-3	将 DAG 从 Airflow 2.x 升级到 3.x
managing-astro-local-env	启动、停止或排查本地 Airflow
setting-up-astro-project	初始化新的 Astro/Airflow 项目
airflow-state-store	按任务进行检查点保存、水位线管理、崩溃安全型 Operator（Airflow 3.3+）
airflow-hitl	暂停 DAG，以等待人工审批或输入（Airflow 3.1+）