Cloud Composer (Managed Apache Airflow)

# Cloud Composer DAG — orchestrate the full GCP pipeline
# File: dags/daily_sales_pipeline.py
# This DAG runs daily, coordinates GCS → Dataflow → BigQuery

from airflow import DAG
from airflow.providers.google.cloud.operators.dataflow import DataflowStartFlexTemplateOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor
from airflow.utils.dates import days_ago
from datetime import timedelta

default_args = {
    'owner': 'data-engineering',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['de-team@yourcompany.com'],
}

with DAG(
    dag_id='daily_sales_pipeline',
    default_args=default_args,
    schedule_interval='0 3 * * *',   # Run every day at 3am UTC
    start_date=days_ago(1),
    catchup=False,
    tags=['sales', 'gcp', 'daily'],
) as dag:

    # Step 1: Wait for source file to land in GCS Bronze
    wait_for_source_file = GCSObjectExistenceSensor(
        task_id='wait_for_source_file',
        bucket='your-bucket',
        object='bronze/sales/{{ ds }}/sales_raw.csv',  # ds = execution date YYYY-MM-DD
        timeout=3600,  # Wait up to 1 hour
        poke_interval=120,
    )

    # Step 2: Run Dataflow job to transform Bronze → Silver in GCS
    run_dataflow_transform = DataflowStartFlexTemplateOperator(
        task_id='run_dataflow_transform',
        project_id='your-project',
        location='us-central1',
        body={
            'launchParameter': {
                'jobName': 'bronze-to-silver-{{ ds_nodash }}',
                'containerSpecGcsPath': 'gs://your-bucket/templates/bronze_to_silver',
                'parameters': { 'run_date': '{{ ds }}' },
            }
        },
    )

    # Step 3: Load Silver data from GCS into BigQuery Gold table
    load_to_bigquery = BigQueryInsertJobOperator(
        task_id='load_to_bigquery',
        configuration={
            'load': {
                'sourceUris': ['gs://your-bucket/silver/sales/{{ ds }}/*.parquet'],
                'destinationTable': { 'projectId': 'your-project', 'datasetId': 'gold', 'tableId': 'daily_sales_summary' },
                'sourceFormat': 'PARQUET',
                'writeDisposition': 'WRITE_APPEND',
                'timePartitioning': { 'field': 'order_date', 'type': 'DAY' },
            }
        },
    )

    # Step 4: Run aggregation query to refresh Gold summary table
    refresh_gold_summary = BigQueryInsertJobOperator(
        task_id='refresh_gold_summary',
        configuration={
            'query': {
                'query': '''
                    INSERT INTO `your_project.gold.regional_performance`
                    SELECT region, SUM(total_revenue) as revenue, COUNT(*) as orders,
                           DATE('{{ ds }}') as summary_date
                    FROM `your_project.gold.daily_sales_summary`
                    WHERE order_date = '{{ ds }}'
                    GROUP BY region
                ''',
                'useLegacySql': False,
            }
        },
    )

    # Define execution order with >> operator
    wait_for_source_file >> run_dataflow_transform >> load_to_bigquery >> refresh_gold_summary