GCP Apache Airflow Managed Service: Orquestación Empresarial en Google Cloud Composer

"""
Pipeline de Producción en GCP Cloud Composer 2/3
Patrón: Deferrable BigQuery ETL + TaskFlow Decorators
Autor: Eduardo Martínez Agrelo (AI & Data Architect)
"""

from __future__ import annotations
from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator

PROJECT_ID = "enterprise-data-mesh-prod"
DATASET_ID = "analytics_staging"
RAW_BUCKET = "gs://raw-telemetry-lakehouse-prod"

DEFAULT_ARGS = {
    "owner": "data-architecture-team",
    "depends_on_past": False,
    "email_on_failure": True,
    "email": ["data-alerts@enterprise.com"],
    "retries": 2,
    "retry_delay": timedelta(minutes=3),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=15),
}

@dag(
    dag_id="gcp_composer_optimized_elt",
    default_args=DEFAULT_ARGS,
    schedule_interval="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=["finops", "bigquery", "deferrable", "production"],
)
def gcp_managed_pipeline():

    # 1. Ingestión GCS a BigQuery Staging (Idempotente)
    ingest_gcs_to_stg = GCSToBigQueryOperator(
        task_id="ingest_telemetry_to_staging",
        bucket=RAW_BUCKET.replace("gs://", ""),
        source_objects=["events/{{ ds }}/*.parquet"],
        destination_project_dataset_table=f"{PROJECT_ID}.{DATASET_ID}.stg_events_{{{{ ds_nodash }}}}",
        source_format="PARQUET",
        write_disposition="WRITE_TRUNCATE",
        autodetect=True,
    )

    # 2. Transformación SQL Serverless (Operador Deferrable)
    # Al activar deferrable=True, el worker cede el hilo al Triggerer
    transform_aggregation = BigQueryInsertJobOperator(
        task_id="transform_analytics_mart",
        configuration={
            "query": {
                "query": f"""
                    CREATE OR REPLACE TABLE `{PROJECT_ID}.analytics_mart.daily_kpis_{{{{ ds_nodash }}}}` AS
                    SELECT 
                        event_type,
                        COUNT(1) AS total_events,
                        AVG(session_duration_sec) AS avg_duration,
                        CURRENT_TIMESTAMP() AS processed_at
                    FROM `{PROJECT_ID}.{DATASET_ID}.stg_events_{{{{ ds_nodash }}}}`
                    GROUP BY 1;
                """,
                "useLegacySql": False,
            }
        },
        deferrable=True,  # Clave FinOps: libera slots de Celery durante el query
    )

    # 3. Validación de Calidad de Datos vía TaskFlow
    @task
    def validate_mart_quality(table_suffix: str) -> dict:
        # Import local dentro de la tarea para evitar degradación de parsing del Scheduler
        from google.cloud import bigquery
        
        client = bigquery.Client()
        query = f"SELECT count(*) as rows_count FROM `{PROJECT_ID}.analytics_mart.daily_kpis_{table_suffix}`"
        query_job = client.query(query)
        result = list(query_job.result())
        count = result[0].rows_count
        
        if count == 0:
            raise ValueError("Data Quality Check Failed: Tabla de KPIs vacía.")
            
        return {"status": "SUCCESS", "row_count": count}

    quality_check = validate_mart_quality(table_suffix="{{ ds_nodash }}")

    # Dependencias declarativas
    ingest_gcs_to_stg >> transform_aggregation >> quality_check

pipeline = gcp_managed_pipeline()