NutriMetrics — Arquitectura de Pipeline de Datos

Filas diarias

~3M

eventos procesados/día

Tablas fuente

PostgreSQL prod

Tareas Airflow

extract + load + QA

SLA pipeline

<45m

05:00 → 05:45 UTC

Retries config

2×5m

por tarea fallida

Flujo del Pipeline — DAG: nutrimetrics_daily_etl cron: 0 5 * * *

💾 PostgreSQL Prod

orders (2.8M)

order_items (9.1M)

menus (12K)

restaurants (850)

📤 Extract (PythonOp)

extract_orders

extract_order_items

extract_menus

extract_restaurants

▲ Load to Snowflake

load_orders

load_order_items

load_menus

load_restaurants

✓ Quality Check

quality_check

⚙ dbt Transform

dbt_run (fct/dim)

dbt_test

📊 Serve / BI

Looker / Tableau

🎯 Decision Framework: Batch vs Streaming

Criterio	Batch	Streaming	✓
Latencia requerida	Horas	Segundos	Batch
Reportes BI	T+1 día OK	Sobreingeniería	Batch
Volumen diario	<1TB → dbt	Streaming	Batch
Coste infra	Bajo	Alto (+Kafka)	Batch
Reprocessing	Nativo (Airflow)	Complejo	Batch

Veredicto: Arquitectura Batch recomendada. Kappa sería prematura para 3M eventos/día con latencia de horas. Revisar en 12 meses si se requiere analítica <5min.

📂 Tablas Fuente — PostgreSQL

orders

2,800,000 filas

incremental: created_at

ACTIVA

order_items

9,100,000 filas

incremental: created_at

ACTIVA

menus

12,000 filas

incremental: updated_at

ACTIVA

restaurants

850 filas

incremental: updated_at

ACTIVA

Destino Snowflake — Warehouse vs Lakehouse

Aspecto	Warehouse ✓	Lakehouse
BI / SQL analytics	Excelente	Bueno
Coste almacenamiento	Aceptable	check Más bajo
ML / unstructured	Limitado	Nativo
Madurez ecosistema	BI tools maduro	Creciendo

⭐ Modelo Dimensional — Star Schema (Snowflake)

fct_orders

FACT TABLE

order_id PK
restaurant_key FK
menu_key FK
user_key FK
date_key FK
total_amount
item_count
created_at

dim_date

date_key PK
full_date · year
month · weekday
is_weekend

dim_menus

menu_key PK
menu_id · name
category
calories · price

dim_users

user_key PK
user_id · email
plan_tier
created_at

dim_restaurants

restaurant_key PK
restaurant_id · name
city · country
cuisine_type

SCD Type 2 en dim_restaurants · Particionado por date_key · Clustering por restaurant_key

✅ Quality Checks — Great Expectations + dbt tests

✓

Completitud orders

order_id, restaurant_id, total_amount NOT NULL

Umbral: 99% · Actual: 98.7%

🔑

Unicidad order_id

Clave primaria sin duplicados

100% unique · 0 duplicados hoy

🕑

Frescura datos

max(created_at) < 26h desde NOW()

Último registro: hace 4.2h ✓

📈

Rango total_amount

Valores entre 0.01€ y 9,999€

3 anomalías detectadas → quarantine

🔗

Integridad referencial

order_items.order_id → orders.order_id

FK constraint validated ✓

📊

Volumen mínimo

Ordenes diarias > 500 (alerta si baja)

Hoy: 8,432 órdenes ✓

🔔 Alertas configuradas

Slack #data-alerts Email: data-team@nutrimetrics.io Datadog APM PagerDuty P2

📄 DAG Generado — nutrimetrics_daily_etl.py

"""Auto-generated Airflow DAG — Pipeline Orchestrator"""

from airflow import DAG
from airflow.utils.dates import days_ago
from datetime import datetime, timedelta
from airflow.operators.python import PythonOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='nutrimetrics_daily_etl',
    schedule_interval='0 5 * * *',
    start_date=days_ago(1),
    catchup=False,
    max_active_runs=1,
    tags=['etl', 'postgres', 'snowflake'],
) as dag:

    @task(retries=2, retry_delay=timedelta(minutes=5))
    def extract_orders(**kwargs):
        execution_date = kwargs.get('ds')
        # Incremental: WHERE created_at > {{ prev_ds }}
        return pg_hook.get_pandas_df(sql, params={'dt': execution_date})

    # ... extract_order_items, extract_menus, extract_restaurants ...

    # Task dependencies
    extract_orders >> load_orders
    extract_order_items >> load_order_items
    load_orders >> quality_check
    load_order_items >> quality_check
    load_menus >> quality_check
    load_restaurants >> quality_check

🕐 Ejecución estimada — Hoy 2026-06-12

05:00 UTC

DAG triggered (schedule)

nutrimetrics_daily_etl · date=2026-06-11

Duración estimada: < 1s

05:00-08m

Extract phase (paralelo x4)

extract_orders · extract_order_items · extract_menus · extract_restaurants

~8 min · 9.1M + 2.8M filas incrementales

05:08-22m

Load to Snowflake (paralelo x4)

COPY INTO raw.orders · raw.order_items · raw.menus · raw.restaurants

~14 min · APPEND mode · staging S3 → Snowflake

05:22-25m

quality_check

Great Expectations: 6 checks · umbral 99%

~3 min · Alertas Slack si fallo

05:25-40m

dbt run + dbt test

stg_orders · fct_orders · dim_restaurants · dim_menus · dim_users

~15 min · 12 modelos · 24 tests

05:40 UTC

✓ Pipeline completado

Snowflake listo para BI · SLA: 40min (<45min objetivo)

Notificación Slack: #data-ops

🔨 Stack Tecnológico Recomendado

Orquestación: Airflow 2.7

Transform: dbt Core 1.7

Destino: Snowflake

Calidad: Great Expectations

Staging: S3 → Snowflake COPY

Monitorización: Datadog + Slack

🚀 DataOps — CI/CD y Observabilidad del Pipeline

🔧 CI/CD Data Pipeline

• GitHub Actions: lint + test en PR
• dbt slim CI: solo modelos afectados
• Pre-commit: sqlfluff + yamllint
• Staging env: datos del último día
• Prod deploy: merge a main

🔍 Observabilidad

• Datadog: métricas Airflow + Snowflake
• Lineage: dbt docs generate
• Anomaly detection: volumen ±30%
• SLA tracking: duración por tarea
• Data catalog: dbt + Atlan (futuro)

🔒 Seguridad y Gobierno

• Secrets: AWS Secrets Manager
• RBAC: roles por dominio en Snowflake
• PII masking: email, phone en staging
• Audit log: Snowflake ACCESS_HISTORY
• Retención: raw 90d, curated 2y

📊 Próximos pasos (Q3 2026)

• Streaming tier: Kafka + Flink para alertas <5min
• ML feature store: Feast
• Reverse ETL: Hightouch → CRM
• Data contracts: OpenAPI schemas
• Costo: Snowflake auto-suspend