Pipeline de Datos en GCP (Airflow + dbt + BigQuery)

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from datetime import datetime

def fetch_api(**ctx):
    # llamada a API y subida a GCS / carga staging
    pass

with DAG("gcp_demo_pipeline",
         start_date=datetime(2025,8,1),
         schedule_interval="0 * * * *",  # cada hora
         catchup=False) as dag:

    extract = PythonOperator(
        task_id="extract_api",
        python_callable=fetch_api,
    )

    transform_sql = BigQueryInsertJobOperator(
        task_id="transform_sql",
        configuration={
            "query": {
              "query": "CREATE OR REPLACE TABLE ds.stg_orders AS SELECT * FROM ds.raw_orders",
              "useLegacySql": false
            }
        }
    )

    extract >> transform_sql

Plantilla mínima: extraer y transformar en BigQuery.

# models/stg_orders.sql (dbt)
{{ config(materialized="view") }}
SELECT
  order_id,
  user_id,
  CAST(total AS NUMERIC) AS total,
  TIMESTAMP(order_ts)   AS order_ts
FROM {{ source("raw", "orders") }}
WHERE status = 'PAID'

Vista staging normalizada y tipada.

# models/sources.yml (dbt)
version: 2
sources:
  - name: raw
    schema: raw
    tables:
      - name: orders
        loaded_at_field: _ingested_at
        freshness:
          warn_after: {count: 2, period: hour}
          error_after: {count: 6, period: hour}

Declaración de fuentes + freshness checks.

# models/business_orders.yml (dbt)
version: 2
models:
  - name: business_orders
    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [order_id]
    columns:
      - name: order_id
        tests: [not_null]

Calidad básica: not_null/unique/relationships.

# Airflow: ejecutar dbt en contenedor (Composer 2)
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

dbt_run = KubernetesPodOperator(
  task_id="dbt_run",
  name="dbt-run",
  namespace="composer",             # o el tuyo
  image="gcr.io/tu-proyecto/dbt-bq:latest",
  cmds=["bash","-lc"],
  arguments=["dbt deps && dbt run --profiles-dir . --target prod"],
  env_vars={"DBT_ENV_SECRET_GCP_SERVICE_ACCOUNT": "{{ var.value.gcp_sa_json }}"},
)

Aísla dependencias y escala por tarea.

# Python → Pandas → GCS → BigQuery (en tarea de Airflow)
import pandas as pd
from google.cloud import storage, bigquery

def transform_and_load(**ctx):
    df = pd.read_json("https://api.example.com/orders")
    df["total"] = df["total"] * 100
    # 1) subir a GCS (parquet)
    df.to_parquet("/tmp/orders.parquet", index=False)
    storage.Client().bucket("mi-bucket").blob("staging/orders.parquet").upload_from_filename("/tmp/orders.parquet")
    # 2) cargar en BigQuery
    bq = bigquery.Client()
    job = bq.load_table_from_uri(
        "gs://mi-bucket/staging/orders.parquet",
        "raw.orders",
        job_config=bigquery.LoadJobConfig(source_format=bigquery.SourceFormat.PARQUET, write_disposition="WRITE_TRUNCATE"),
    )
    job.result()

Ruta común para APIs: pandas → parquet → load job.

-- BigQuery: capa de negocio
CREATE OR REPLACE TABLE mart.business_orders
PARTITION BY DATE(order_ts)
CLUSTER BY user_id AS
SELECT
  o.order_id, u.country, o.total, o.order_ts
FROM mart.stg_orders o
LEFT JOIN mart.dim_users u USING (user_id);

Particiona y clusteriza para costo/velocidad.

# dag_airflow_etl.py — igual que el diagrama (1 BQ → 2 Python → 3 BQ)
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from google.cloud import bigquery, storage
import pandas as pd

PROJECT_ID = "{{ var.value.gcp_project_id }}"
BUCKET     = "mi-bucket-etl"
RAW_DS     = "dataset"      # origen (dataset.source)
WORK_DS    = "work"         # staging temporal
TARGET_DS  = "dataset"      # destino (dataset.target)
TARGET_TBL = "target"

with DAG(
    dag_id="etl_bq_python_bq",
    start_date=datetime(2025, 8, 1),
    schedule="@daily",
    catchup=False,
    default_args={"retries": 2},
    tags=["gcp","airflow","bigquery"],
) as dag:

    # (1) BigQueryOperator — SELECT * FROM dataset.source -> work.stg_source
    bq_extract = BigQueryInsertJobOperator(
        task_id="extract_from_source",
        location="US",
        configuration={
            "query": {
                "query": f"""
                    CREATE OR REPLACE TABLE '{PROJECT_ID}.{WORK_DS}.stg_source' AS
                    SELECT * FROM '{PROJECT_ID}.{RAW_DS}.source'
                """,
                "useLegacySql": False,
            }
        },
    )

    # (2) PythonOperator — procesamos con pandas y guardamos Parquet en GCS
    def process_with_pandas(**context):
        client = bigquery.Client(project=PROJECT_ID)
        df = client.query(
            f"SELECT * FROM '{PROJECT_ID}.{WORK_DS}.stg_source'"
        ).result().to_dataframe()

        if "total" in df.columns:
            df["total"] = df["total"].fillna(0) * 100

        uri = f"gs://{BUCKET}/etl_tmp/processed_{{context['ds_nodash']}}.parquet"
        tmp = "/tmp/processed.parquet"
        df.to_parquet(tmp, index=False)
        storage.Client().bucket(BUCKET).blob(
            f"etl_tmp/processed_{context['ds_nodash']}.parquet"
        ).upload_from_filename(tmp)
        return uri

    py_transform = PythonOperator(
        task_id="process_python",
        python_callable=process_with_pandas,
    )

    # (3) BigQueryInsertJobOperator — cargamos el Parquet a dataset.target
    bq_load = BigQueryInsertJobOperator(
        task_id="load_to_target",
        location="US",
        configuration={
            "load": {
                "sourceUris": ["{{ ti.xcom_pull(task_ids='process_python') }}"],
                "destinationTable": {
                    "projectId": PROJECT_ID,
                    "datasetId": TARGET_DS,
                    "tableId": TARGET_TBL,
                },
                "sourceFormat": "PARQUET",
                "writeDisposition": "WRITE_TRUNCATE",
            }
        },
    )

    bq_extract >> py_transform >> bq_load

Replica 1) BigQueryOperator → 2) PythonOperator → 3) BigQueryInsertJobOperator del diagrama.