Back to Details

managed-airflow-dag-authoring

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

GCP Managed Airflow DAG Authoring Guide

GCP托管式Airflow DAG编写指南

This skill guides you through authoring and validating Apache Airflow DAGs for Managed Service for Apache Airflow (MSAA; formerly Cloud Composer) environments.
本技能将指导你为托管式Apache Airflow服务（MSAA；前身为Cloud Composer）环境编写并验证Apache Airflow DAG。

Phase 1: Context Discovery

第一阶段：上下文探查

Before writing any DAG code, you MUST understand the constraints (e.g. version of Airflow) and capabilities of your target environment if user is willing to provide them.
在编写任何DAG代码之前，若用户愿意提供相关信息，你必须了解目标环境的约束条件（如Airflow版本）和功能特性。

1.1 Identify Target Environment & Access

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

Determine if you have direct access to the target Managed Airflow environment, local development environment or if you are working offline (only changing local files without validation).
  • If environment access is available: Use 
    gcloud
    to inspect the environment (see Section 1.3).
  • If offline: Rely on user provided details.
确认你是否能直接访问目标托管式Airflow环境、本地开发环境，或者是否处于离线工作状态（仅修改本地文件而不进行验证）。
  • 若可访问环境： 使用
    gcloud
    工具探查环境（参见1.3节）。
  • 若离线： 依赖用户提供的详细信息。

1.2 Identify Development Environment

1.2 确定开发环境

Determine if a local development environment is available.
  • Check if 
    composer-dev
    CLI is installed.
  • Check if a local Python environment with 
    airflow
    is available.
确认是否有可用的本地开发环境。
  • 检查是否已安装
    composer-dev
    CLI工具。
  • 检查是否有包含
    airflow
    的本地Python环境。

1.3 Inspect Target Environment (if available and requested)

1.3 探查目标环境（若可用且用户要求）

Run the following commands to discover version constraints:
  1. Get Airflow/Image Version:
    bash
    gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.softwareConfig.imageVersion)"
  2. Get Installed Packages (Versions):
    bash
    gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.softwareConfig.pypiPackages)"
  3. Get DAGs GCS Bucket:
    bash
    gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.dagGcsPrefix)"
运行以下命令来发现版本约束：
  1. 获取Airflow/镜像版本：
    bash
    gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.softwareConfig.imageVersion)"
  2. 获取已安装包（版本）：
    bash
    gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.softwareConfig.pypiPackages)"
  3. 获取DAG的GCS存储桶：
    bash
    gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.dagGcsPrefix)"

Phase 2: DAG Authoring Best Practices

第二阶段：DAG编写最佳实践

2.1 General Airflow Best Practices

2.1 Airflow通用最佳实践

  • Idempotency: Every task SHOULD be idempotent. Running it multiple times with the same inputs (e.g., execution date) SHOULD produce the same result and not duplicate data.
  • No Top-Level Code Execution: Do NOT execute database queries, external API calls, or heavy computations at the top level of the DAG file (outside of tasks/operators). This code runs every few seconds during DAG parsing and will degrade performance.
  • Explicit Catchup: Always set 
    catchup=False
    in the DAG definition unless historical backfilling is explicitly required.
  • Use Airflow Variables/Connections: Never hardcode credentials or environment-specific configs. Use 
    Variable.get()
    (with 
    deserialize_json=True
    if applicable) and 
    BaseHook.get_connection()
    . Access variables via Jinja templates (e.g., 
    {{ var.value.my_var }}
    ) to avoid database calls during DAG parsing.
  • 幂等性： 每个任务都应具备幂等性。使用相同输入（如执行日期）多次运行任务，应产生相同结果且不会重复数据。
  • 避免顶层代码执行： 不要在DAG文件的顶层（任务/操作符之外）执行数据库查询、外部API调用或密集计算。这段代码会在DAG解析时每隔几秒运行一次，会降低性能。
  • 显式设置Catchup： 除非明确需要历史数据回填，否则在DAG定义中始终设置
    catchup=False
    。
  • 使用Airflow变量/连接： 切勿硬编码凭证或环境特定配置。使用
    Variable.get()
    （若适用可搭配
    deserialize_json=True
    ）和
    BaseHook.get_connection()
    。通过Jinja模板访问变量（如
    {{ var.value.my_var }}
    ），避免DAG解析时调用数据库。

2.2 Airflow 2 vs Airflow 3 Compatibility

2.2 Airflow 2与Airflow 3兼容性

Use managed-airflow-migrations skill to navigate adjusting the code to specific target Airflow version.
使用managed-airflow-migrations技能来调整代码以适配特定目标Airflow版本。

Phase 3: Validation Process

第三阶段：验证流程

You MUST validate DAGs before concluding your task.
在完成任务前，你必须对DAG进行验证。

3.1 Local Validation (Offline/Pre-deployment)

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

3.1.1 Static Analysis & Linting

3.1.1 静态分析与代码检查

Use 
ruff
or 
pylint
if available.
bash
ruff check path/to/dag.py
  • If targeting Airflow 3, check with Airflow 3 rules if rulesets are available.
若可用，使用
ruff
或
pylint
工具。
bash
ruff check path/to/dag.py
  • 若目标为Airflow 3，若有可用规则集，请使用Airflow 3规则进行检查。

3.1.2 Local Dev Environment (
composer-dev
)

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

If the user has 
composer-dev
configured:
  1. Copy the DAG to the local directory with DAGs:
    bash
    cp path/to/dag.py $(composer-dev describe {local_env} --format="value(dags_directory)")
  2. Verify parsing:
    bash
    composer-dev run-airflow-cmd {local_env} dags list-import-errors
若用户已配置
composer-dev
：
  1. 将DAG复制到本地DAG目录：
    bash
    cp path/to/dag.py $(composer-dev describe {local_env} --format="value(dags_directory)")
  2. 验证解析情况：
    bash
    composer-dev run-airflow-cmd {local_env} dags list-import-errors

3.2: Target Environment Validation

3.2: 目标环境验证

Only perform these steps if you have GCP access and are authorized to deploy to a target environment.
仅当你拥有GCP访问权限且获许部署到目标环境时，执行以下步骤。

3.2.1 Deploy to GCS

3.2.1 部署到GCS

Upload the DAG to the target environment's GCS bucket:
bash
gcloud storage cp path/to/dag.py gs://{target_bucket}/dags/
将DAG上传至目标环境的GCS存储桶：
bash
gcloud storage cp path/to/dag.py gs://{target_bucket}/dags/

3.2.2 Verify via Airflow CLI

3.2.2 通过Airflow CLI验证

Wait 1-2 minutes for the scheduler to parse the file, then run:
  1. Check for Import Errors:
    bash
    gcloud composer environments run {env_name} \
    --location {region} \
    dags list-import-errors
Pass Criteria: Output should be "No data found" or empty.
  1. Verify DAG is Listed:
    bash
    gcloud composer environments run {env_name} \
    --location {region} \
    dags list | grep {dag_id}
等待1-2分钟让调度器解析文件，然后运行：
  1. 检查导入错误：
    bash
    gcloud composer environments run {env_name} \
    --location {region} \
    dags list-import-errors
通过标准： 输出应为“No data found”或空内容。
  1. 验证DAG已被列出：
    bash
    gcloud composer environments run {env_name} \
    --location {region} \
    dags list | grep {dag_id}

3.2.3 Monitor Cloud Logging

3.2.3 监控Cloud Logging

Check for runtime parsing errors in Cloud Logging:
query
resource.type="cloud_composer_environment"
resource.labels.environment_name="{env_name}"
log_id("airflow-scheduler")
severity>=ERROR
textPayload:"{dag_file_name}"
在Cloud Logging中检查运行时解析错误：
query
resource.type="cloud_composer_environment"
resource.labels.environment_name="{env_name}"
log_id("airflow-scheduler")
severity>=ERROR
textPayload:"{dag_file_name}"

Definition of Done

完成标准

  • DAG code adheres to Airflow version constraints of the target environment.
  • DAG code follows best practices (no top-level execution, idempotent if possible).
  • DAG parses locally without import errors.
  • (If environment is available) DAG is deployed to the target environment and verified to have no import errors.
  • DAG代码符合目标环境的Airflow版本约束。
  • DAG代码遵循最佳实践（无顶层代码执行、尽可能具备幂等性）。
  • DAG在本地解析无导入错误。
  • （若环境可用）DAG已部署到目标环境并验证无导入错误。