managed-airflow-dag-authoring

GCP Managed Airflow DAG Authoring Guide

This skill guides you through authoring and validating Apache Airflow DAGs for

Managed Service for Apache Airflow (MSAA; formerly Cloud Composer) environments.

--------------------------------------------------------------------------------

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.

1.1 Identify Target Environment & Access

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.

1.2 Identify Development Environment

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.

1.3 Inspect Target Environment (if available and requested)

Run the following commands to discover version constraints:

1. Get Airflow/Image Version:

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

1. Get Installed Packages (Versions):

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

1. Get DAGs GCS Bucket:

    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

Use managed-airflow-migrations skill to navigate adjusting the code to

specific target Airflow version.

--------------------------------------------------------------------------------

Phase 3: Validation Process

You MUST validate DAGs before concluding your task.

3.1 Local Validation (Offline/Pre-deployment)

3.1.1 Static Analysis & Linting

Use ruff or pylint if available.

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)

If the user has composer-dev configured:

1. Copy the DAG to the local directory with DAGs:

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

1. Verify parsing:

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

3.2: Target Environment Validation

Only perform these steps if you have GCP access and are authorized to deploy to

a target environment.

3.2.1 Deploy to GCS

Upload the DAG to the target environment's GCS bucket:

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

3.2.2 Verify via Airflow CLI

Wait 1-2 minutes for the scheduler to parse the file, then run:

1. Check for Import Errors:

    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:

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

3.2.3 Monitor Cloud Logging

Check for runtime parsing errors in 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.