managed-airflow-dag-authoring

GCP Managed Airflow DAG 编写指南

本技能将指导你为 Managed Service for Apache Airflow（MSAA；前身为 Cloud Composer）环境编写和验证 Apache Airflow DAG。

---

署名

必须阅读：执行任何 gcloud、REST API 或 MCP 命令之前，你必须阅读使用归属信息。直接在终端中执行命令时（绝不可在生成的脚本、Terraform 或用户代码中），请在 gcloud 前加上 CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:managed-airflow-dag-authoring)" ，并将 agent-skills/1.0 (skill:managed-airflow-dag-authoring) 添加到 User-Agent 标头中。

阶段 1：了解上下文

编写任何 DAG 代码之前，如果用户愿意提供相关信息，你必须了解目标环境的限制（例如 Airflow 版本）和功能。

1.1 确定目标环境和访问权限

确定你是否能直接访问目标 Managed Airflow 环境，是否有本地开发环境，或者你是否处于离线状态（只能修改本地文件，无法进行验证）。

• 如果可以访问环境： 使用 gcloud 检查环境（参见第 1.3 节）。
• 如果处于离线状态： 根据用户提供的详细信息进行操作。

1.2 确定开发环境

确定是否有可用的本地开发环境。

• 检查是否已安装 composer-dev CLI。
• 检查是否有包含 airflow 的本地 Python 环境。

1.3 检查目标环境（如果可用且用户要求）

运行以下命令以了解版本限制：

1. 获取 Airflow/映像版本：

   gcloud composer environments describe {env_name} \
       --location {region} \
       --format="value(config.softwareConfig.imageVersion)"
   

2. 获取已安装的软件包（版本）：

   gcloud composer environments describe {env_name} \
       --location {region} \
       --format="value(config.softwareConfig.pypiPackages)"
   

3. 获取 DAG 的 GCS 存储桶：

   gcloud composer environments describe {env_name} \
       --location {region} \
       --format="value(config.dagGcsPrefix)"
   

---

阶段 2：DAG 编写最佳实践

2.1 Airflow 通用最佳实践

• 幂等性： 每个任务都应当具有幂等性。使用相同的输入（例如执行日期）多次运行时，应当产生相同的结果，且不重复数据。
• 不要在顶层执行代码： 不要在 DAG 文件顶层（任务/运算符之外）执行数据库查询、外部 API 调用或复杂计算。DAG 解析期间每隔几秒就会运行这些代码，从而降低性能。
• 显式设置补运行： 除非明确要求回补历史数据，否则始终在 DAG 定义中设置 catchup=False。
• 使用 Airflow 变量/连接： 切勿硬编码凭据或特定于环境的配置。请使用 Variable.get()（如果适用，设置 deserialize_json=True）和 BaseHook.get_connection()。通过 Jinja 模板访问变量（例如 {{ var.value.my_var }}），避免在 DAG 解析期间调用数据库。

2.2 Airflow 2 与 Airflow 3 的兼容性

使用 managed-airflow-migrations 技能，根据目标 Airflow 版本调整代码。

---

阶段 3：验证流程

结束任务之前，你必须验证 DAG。

3.1 本地验证（离线/部署前）

3.1.1 静态分析和 Lint

如果可用，请使用 ruff 或 pylint。

ruff check path/to/dag.py

• 如果目标是 Airflow 3，并且有相应规则集，请使用 Airflow 3 规则进行检查。

3.1.2 本地开发环境（composer-dev）

如果用户已配置 composer-dev：

1. 将 DAG 复制到包含 DAG 的本地目录：

   cp path/to/dag.py $(composer-dev describe {local_env} --format="value(dags_directory)")
   

2. 验证解析：

   composer-dev run-airflow-cmd {local_env} dags list-import-errors
   

3.2：目标环境验证

仅在你有 GCP 访问权限且获准部署到目标环境时，执行以下步骤。

3.2.1 部署到 GCS

将 DAG 上传到目标环境的 GCS 存储桶：

gcloud storage cp path/to/dag.py gs://{target_bucket}/dags/

3.2.2 通过 Airflow CLI 验证

等待 1-2 分钟，让调度器解析文件，然后运行：

1. 检查导入错误：

   gcloud composer environments run {env_name} \
       --location {region} \
       dags list-import-errors
   

通过标准： 输出应为“No data found”或为空。

2. 确认 DAG 已列出：

   gcloud composer environments run {env_name} \
       --location {region} \
       dags list | grep {dag_id}
   

3.2.3 监控 Cloud Logging

在 Cloud Logging 中检查运行时解析错误：

resource.type="cloud_composer_environment"
resource.labels.environment_name="{env_name}"
log_id("airflow-scheduler")
severity>=ERROR
textPayload:"{dag_file_name}"

---

完成标准

• DAG 代码符合目标环境的 Airflow 版本限制。
• DAG 代码遵循最佳实践（不在顶层执行；如有可能，应具备幂等性）。
• DAG 可在本地解析，且没有导入错误。
• （如果环境可用）DAG 已部署到目标环境，并已确认没有导入错误。