GCP Managed Airflow DAG Authoring Guide

Phase 1: Context Discovery

1.1 Identify Target Environment & Access

• If environment access is available: Use

  gcloud

  to inspect the environment (see Section 1.3).
• If offline: Rely on user provided details.

1.2 Identify Development Environment

• Check if

  composer-dev

  CLI is installed.
• Check if a local Python environment with

  airflow

  is available.

1.3 Inspect Target Environment (if available and requested)

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)"

Phase 2: DAG Authoring Best Practices

2.1 General Airflow Best Practices

• 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.

2.2 Airflow 2 vs Airflow 3 Compatibility

Phase 3: Validation Process

3.1 Local Validation (Offline/Pre-deployment)

3.1.1 Static Analysis & Linting

ruff

pylint

ruff check path/to/dag.py

• If targeting Airflow 3, check with Airflow 3 rules if rulesets are available.

3.1.2 Local Dev Environment (

composer-dev

)

composer-dev

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

3.2: Target Environment Validation

3.2.1 Deploy to GCS

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

3.2.2 Verify via Airflow CLI

1. Check for Import Errors:
   bash

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

1. Verify DAG is Listed:
   bash

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

3.2.3 Monitor Cloud Logging

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.