Google Cloud — Composer (Workflow Orchestration)

Cloud Composer is a managed workflow orchestration service built on Apache Airflow. It automates the creation, scheduling, and monitoring of complex data pipelines across GCP services. CI/CD integration lets you manage DAGs as code alongside your application deployments using source-controlled repositories and automated sync.

Why Cloud Composer Exists and When to Use It

Cloud Composer is Google Cloud's managed Apache Airflow service. It exists because orchestrating complex data pipelines manually is error-prone and unscalable. You should use Composer when you have workflows that span multiple services (e.g., BigQuery, Cloud Storage, Pub/Sub) and require scheduling, retries, and monitoring. It's not for simple cron jobs—use Cloud Scheduler for those. Composer shines when you need to coordinate tasks with dependencies, handle failures gracefully, and audit execution history. In production, we've seen teams misuse Composer for real-time processing; it's designed for batch workflows with minutes-to-hours latency. If you need sub-second orchestration, look at Cloud Workflows or Pub/Sub directly.

airflow_variables:
  - key: project_id
    value: my-gcp-project
  - key: dataset_id
    value: analytics
  - key: bucket_name
    value: my-data-lake

Architecture: Components and Their Roles

A Cloud Composer environment consists of several GCP resources: a Cloud Storage bucket for DAGs and logs, a Cloud SQL instance for the Airflow metadata database, and a GKE cluster (or Cloud Composer 2's autopilot) running Airflow components (scheduler, workers, web server). The scheduler parses DAGs and triggers tasks; workers execute them; the web server provides the UI. The metadata database tracks task states, DAG runs, and variables. In production, the GKE cluster is the most failure-prone component—node pool autoscaling can cause delays if not configured correctly. Always set minimum node counts to avoid cold starts for critical pipelines.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

dag = DAG(
    'hello_world',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@daily',
    catchup=False
)

task = BashOperator(
    task_id='say_hello',
    bash_command='echo "Hello from Composer"',
    dag=dag
)

DAG Design: Structure, Dependencies, and Best Practices

A DAG (Directed Acyclic Graph) defines your workflow. Each task is an operator (e.g., PythonOperator, BigQueryOperator). Dependencies are set using >> or set_downstream(). Best practices: keep DAGs idempotent—rerunning should produce the same result. Use catchup=False for backfills unless you need historical runs. Set max_active_runs=1 for sequential execution. Avoid putting heavy computation in the scheduler—use PythonOperator with execution_timeout. In production, we structure DAGs with clear task groups and use ShortCircuitOperator to skip downstream tasks based on conditions. Always test DAGs locally with airflow dags test before deploying.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract():
    print("Extracting data...")

def transform():
    print("Transforming data...")

def load():
    print("Loading data...")

with DAG(
    'etl_pipeline',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@hourly',
    catchup=False,
    max_active_runs=1
) as dag:
    extract_task = PythonOperator(task_id='extract', python_callable=extract)
    transform_task = PythonOperator(task_id='transform', python_callable=transform)
    load_task = PythonOperator(task_id='load', python_callable=load)

    extract_task >> transform_task >> load_task

Connecting to GCP Services: BigQuery, Cloud Storage, and Pub/Sub

Composer integrates natively with GCP via Airflow's GCP operators and hooks. For BigQuery, use BigQueryInsertJobOperator to run queries and BigQueryToGCSOperator to export data. For Cloud Storage, GCSObjectExistenceSensor waits for a file, and GCSToBigQueryOperator loads data. For Pub/Sub, PubSubPullSensor listens for messages. In production, manage connections via Airflow's UI or environment variables—never hardcode credentials. Use service account impersonation for fine-grained access. We've seen failures when operators lack proper IAM permissions; always test with the least privileged service account.

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor
from datetime import datetime

with DAG(
    'gcp_integration',
    start_date=datetime(2025, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    wait_for_file = GCSObjectExistenceSensor(
        task_id='wait_for_file',
        bucket='my-bucket',
        object='data.csv',
        google_cloud_conn_id='gcp_default'
    )
    run_query = BigQueryInsertJobOperator(
        task_id='run_query',
        configuration={
            'query': {
                'query': 'SELECT COUNT(*) FROM `my_project.my_dataset.my_table`',
                'useLegacySql': False
            }
        },
        google_cloud_conn_id='gcp_default'
    )
    wait_for_file >> run_query

Handling Failures: Retries, Alerts, and SLAs

Failures are inevitable. Configure task retries with retries and retry_delay parameters. Use email_on_failure or Slack webhooks for alerts. Set SLAs on tasks to detect slow runs—if a task exceeds its sla, Airflow sends a notification without failing the task. In production, we use on_failure_callback to send alerts to PagerDuty. Avoid infinite retries; set max_retry_delay to cap exponential backoff. For critical pipelines, implement a dead-letter DAG that runs on failure to handle cleanup. We've seen cascading failures when retries overload downstream systems—use retry_exponential_backoff wisely.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def failing_task():
    raise ValueError("Simulated failure")

with DAG(
    'retry_example',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@daily',
    catchup=False,
    default_args={
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
        'email_on_failure': True,
        'email': ['alerts@example.com']
    }
) as dag:
    task = PythonOperator(
        task_id='failing_task',
        python_callable=failing_task
    )

Performance Tuning: Worker Concurrency and Pool Management

Composer performance depends on worker configuration. In Airflow, parallelism controls max task instances across the environment, dag_concurrency controls tasks per DAG, and max_active_tasks_per_dag limits per-DAG concurrency. In production, start with parallelism=32 and adjust based on task duration. Use Airflow pools to limit concurrency for resource-constrained systems (e.g., database connections). For long-running tasks, set execution_timeout to prevent hanging. Monitor worker utilization in GCP console—if workers are idle, reduce node count; if tasks queue, increase parallelism or add nodes. We've seen teams over-provision workers, wasting money. Right-size by analyzing task duration and frequency.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def db_task():
    print("Accessing database...")

with DAG(
    'pool_example',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    task = PythonOperator(
        task_id='db_task',
        python_callable=db_task,
        pool='db_pool',
        pool_slots=1
    )

Security: Secrets, IAM, and Network Controls

Never store secrets in DAG code. Use Airflow's Secrets Backend (e.g., Google Cloud Secret Manager) to retrieve credentials at runtime. For IAM, create a dedicated service account per environment with minimal permissions. Use VPC-SC to restrict network access to GCP services. In production, we encrypt connections with TLS and use private IP for the Composer environment to avoid exposure. Regularly rotate service account keys. We've seen breaches when teams hardcoded API keys in DAGs—always use Secret Manager. Also, enable audit logging to track DAG modifications.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.secrets.secret_manager import CloudSecretManagerBackend
from datetime import datetime

def use_secret():
    from airflow.models import Variable
    api_key = Variable.get("my_api_key")
    print(f"Using API key: {api_key[:4]}...")

with DAG(
    'secret_example',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    task = PythonOperator(
        task_id='use_secret',
        python_callable=use_secret
    )

Monitoring and Logging: What to Watch and How

Monitor Composer at three levels: infrastructure (GKE cluster health, Cloud SQL CPU), Airflow metrics (DAG duration, task failures, scheduler heartbeat), and business metrics (data freshness). Use Cloud Monitoring dashboards for GKE and Cloud SQL. Enable Airflow metrics via StatsD to Cloud Monitoring. For logging, Airflow logs are stored in Cloud Logging; set up log-based metrics for error rates. In production, we alert on scheduler heartbeat misses—if the scheduler stops, no DAGs run. Also monitor task queue depth; a growing queue indicates worker shortage. We've seen silent failures when logs were misconfigured—always test log delivery.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import logging

def monitored_task():
    logging.info("Task started")
    # business logic
    logging.info("Task completed")

with DAG(
    'monitoring_example',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    task = PythonOperator(
        task_id='monitored_task',
        python_callable=monitored_task
    )

CI/CD for DAGs: Testing and Deployment Pipelines

Treat DAGs as code: version control, code review, and automated testing. Use a CI/CD pipeline (Cloud Build, GitHub Actions) to lint DAGs with pylint, run unit tests with pytest, and validate syntax with airflow dags test. Deploy by syncing DAG files to the Composer bucket. In production, we use a staging environment that mirrors production to test DAGs before promotion. Avoid deploying directly to production—use a canary approach. We've seen broken DAGs take down the scheduler; validate DAGs in CI to prevent this. Also, test connections and operators with integration tests against actual GCP services.

steps:
  - name: 'python:3.10'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        pip install apache-airflow
        airflow dags list-import-errors
  - name: 'gcr.io/cloud-builders/gsutil'
    args: ['cp', '-r', 'dags/', 'gs://my-composer-bucket/dags/']

Cost Optimization: Rightsizing and Autoscaling

Composer costs come from GKE nodes, Cloud SQL, and Cloud Storage. To optimize, use preemptible VMs for worker nodes (if tasks are fault-tolerant). Set autoscaling with minimum and maximum node counts based on peak load. Use Cloud SQL with high-availability only for production; dev environments can use single-zone. Delete old DAG runs and logs with airflow db clean. In production, we saved 40% by switching to preemptible nodes and reducing Cloud SQL tier. Monitor cost in GCP billing reports and set budgets. We've seen teams leave idle environments running—use a scheduler to stop non-production environments after hours.

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

with DAG(
    'cleanup_old_runs',
    start_date=datetime(2025, 1, 1),
    schedule_interval='@weekly',
    catchup=False
) as dag:
    clean = BashOperator(
        task_id='clean_db',
        bash_command='airflow db clean --clean-before-timestamp "{{ macros.ds_add(ds, -30) }}" --skip-archives',
        env={"AIRFLOW_HOME": "/home/airflow"}
    )

Advanced Patterns: Dynamic DAGs, Sensors, and SubDAGs

For advanced use cases, generate DAGs dynamically by reading config from Cloud Storage. Use sensors to wait for external events (e.g., file arrival, Pub/Sub message). SubDAGs (or Task Groups) help organize complex workflows. In production, we use dynamic DAGs to create one DAG per customer, each with custom logic. Be careful: too many DAGs can overwhelm the scheduler. Use max_active_runs to limit concurrency. Sensors should have a timeout to avoid hanging. We've seen sensors with no timeout run for days—always set timeout and poke_interval. For long waits, use ExternalTaskSensor to wait for another DAG's completion.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import json

def create_dag(customer_id, schedule):
    with DAG(
        f'customer_{customer_id}',
        start_date=datetime(2025, 1, 1),
        schedule_interval=schedule,
        catchup=False
    ) as dag:
        task = PythonOperator(
            task_id='process',
            python_callable=lambda: print(f"Processing {customer_id}")
        )
    return dag

# Example: read config from GCS
config = [{"id": "123", "schedule": "@daily"}]
for cfg in config:
    globals()[f'dag_{cfg["id"]}'] = create_dag(cfg["id"], cfg["schedule"])

Migration from Airflow on Compute Engine to Composer

Migrating from self-managed Airflow to Composer reduces operational overhead. Steps: export DAGs, connections, and variables. Use airflow export to dump metadata. Recreate connections in Composer via UI or API. Test DAGs in a staging Composer environment. In production, we migrated by running both systems in parallel and switching DNS after validation. Watch out for differences: Composer uses GKE networking, so update firewall rules. Also, Composer's Airflow version may differ—test for deprecated operators. We've seen migration fail due to missing Python packages—use requirements.txt in Composer's PyPI packages.

airflow connections export connections.json
airflow variables export variables.json
airflow dags list -o json > dags_list.json