Cloud Composer Orchestrating Data Workflows

# example_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    "owner": "data_engineer",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

def extract(**kwargs):
    # Example: pull orders from Cloud SQL and store a file in GCS
    # Return a small reference (e.g., path or job id) that downstream tasks can use
    extracted_path = "gs://my-bucket/orders/2023-01-01.json"
    return extracted_path  # returned value becomes an XCom

def validate(**kwargs):
    ti = kwargs["ti"]
    extracted = ti.xcom_pull(task_ids="extract")
    # Validate extracted data
    valid = True
    if not valid:
        raise ValueError("Validation failed")
    return "validated"

def transform(**kwargs):
    ti = kwargs["ti"]
    extracted = ti.xcom_pull(task_ids="extract")
    # Launch Dataflow job or run transformation logic
    transformed_path = "gs://my-bucket/transformed/2023-01-01.parquet"
    return transformed_path

def load(**kwargs):
    ti = kwargs["ti"]
    transformed = ti.xcom_pull(task_ids="transform")
    # Load transformed data into BigQuery
    return "loaded"

with DAG(
    dag_id="ecommerce_etl",
    default_args=default_args,
    description="Daily ETL for e-commerce orders",
    schedule_interval="0 2 * * *",  # daily at 02:00
    start_date=datetime(2023, 1, 1),
    catchup=False,
) as dag:

    t_extract = PythonOperator(task_id="extract", python_callable=extract)
    t_validate = PythonOperator(task_id="validate", python_callable=validate)
    t_transform = PythonOperator(task_id="transform", python_callable=transform)
    t_load = PythonOperator(task_id="load", python_callable=load)

    # Set dependencies (extract -> validate -> transform -> load)
    t_extract >> t_validate >> t_transform >> t_load