Skip to content

Christiano0407/data-platform

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

53 Commits
db
db
 
 
nginx
nginx
 
 
tests
tests
 
 
workers
workers
 
 
.env-example
.env-example
 
 
.gitignore
.gitignore
 
 
README.md
README.md
 
 
docker-compose.yml
docker-compose.yml
 
 

Repository files navigation

Data Pipeline Platform

ETL Pipeline: CSV → MinIO → PostgreSQL. Pipeline de procesamiento de datos para productos con validación, limpieza y carga.

Tabla de Contenidos

  1. Arquitectura
  2. Estructura del Proyecto
  3. Componentes
  4. Guía de Inicio Rápido
  5. Explicación de Logs
  6. Errores Comunes y Soluciones
  7. Variables de Entorno
  8. Comandos Útiles
  9. Diagrama del Flujo

Arquitectura

┌─────────────────────────────────────────────────────────────────────────────────────────────┐
│                        DATA PIPELINE PLATFORM                              │
├─────────────────────────────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   ┌──────────┐                              ┌──────────────────┐       │
│   │   CSV   │                              │  PostgreSQL     │       │
│   │  File   │───────────────────────────▶│  (Catalog)     │       │
│   └──────────┘                              └──────────────────┘       │
│        │                                             ▲                │
│        ▼                                             │                │
│   ┌─────────┐    ┌─────────┐    ┌─────────┐         │                │
│   │   W1   │───▶│   W2   │───▶│   W3   │─────────┘                │
│   │ Ingest │    │ Clean  │    │ Load  │                          │
│   └────┬────┘    └────┬────┘    └────┬────┘                          │
│        │                │                │                              │
│        ▼                │                │                              │
│   ┌──────────┐        │                │                              │
│   │  MinIO  │────────┘                │                              │
│   │  (raw)  │◀─────────────────────┘                              │
│   └────────┘                       │                                        │
│                                ▼                                        │
│                         ┌──────────────┐                                 │
│                         │   MinIO    │                                 │
│                         │ (processed)│◀─────────────┘                              │
│                         └──────────────┘                                           │
│                                                                     │
│   ┌─────────────────────────────────────────┐                              │
│   │  PostgreSQL (Staging - Temporal)          │                              │
│   └─────────────────────────────────────────┘                              │
│                                                                     │
└─────────────────────────────────────────────────────────────────────────────┘

Flujo de Datos

Etapa Worker Descripción Input Output
1 W1 (Ingest) Extrae y carga CSV a storage CSV local MinIO raw/ + staging table
2 W2 (Clean) Valida y normaliza datos Staging table Rows validados/rechazados
3 W3 (Load) Carga a BD y exporta Parquet Rows validados PostgreSQL + MinIO processed/

Estructura del Proyecto

data-platform/
├── .env                          # Variables de entorno (NO commitear)
├── .env.example                  # Template de variables
├── docker-compose.yml             # Orquestación de servicios
├── README.md                   # Este archivo
│
├── nginx/                      # Reverse Proxy
│   ├── Dockerfile
│   ├── nginx.conf
│   └── conf.d/
│       └── api-gateway.conf
│
├── workers/                   # ETL Workers
│   ├── Dockerfile           # Multi-stage build con uv
│   ├── pyproject.toml     # Dependencias Python
│   ├── uv.lock
│   ├── .dockerignore
│   ├── src/workers/
│   │   ├── __init__.py
│   │   ├── core/
│   │   │   ├── config.py       # Settings (Pydantic)
│   │   │   ├── database.py   # SQLAlchemy session
│   │   │   └── logging.py  # Structlog setup
│   │   ├── pipeline/
│   │   │   ├── orchestrator.py    # Orquestador principal
│   │   │   ├── validators/
│   │   │   │   ├── schemas.py   # Modelos Pydantic
│   │   │   │   └── rules.py  # Reglas de negocio
│   │   │   └── workers/
│   │   │       ├── w1_ingest.py   # W1: Ingesta CSV
│   │   │       ├── w2_clean.py   # W2: Limpieza/Validación
│   │   │       └── w3_load.py   # W3: Carga PostgreSQL
│   │   ├── storage/
│   │   │   ├── s3_minio.py      # Cliente MinIO/S3
│   │   │   └── repositories.py  # Repositorios SQLAlchemy
│   │   └── utils/
│   │       ├── constants.py
│   │       ├── exceptions.py
│   │       └── helpers.py
│   └── .venv/                     # Virtual environment local (dev)
│
└── db/
    ├── data/
    │   └── data_api_product_v1.csv    # CSV de productos
    ├── postgres-sql/
    │   └── init.sql                   # Schema inicial
    ├── minio/                        # Scripts MinIO
    └── pgadmin/                      # Config pgAdmin

Componentes

Componente Imagen Puertos Descripción
postgres postgres:15 5436 Base de datos relacional
minio minio/minio:latest 9000 (S3), 9001 (Console) Storage S3-compatible
nginx custom 8080 Reverse proxy / API Gateway
workers data-platform-workers (batch) Pipeline ETL

Servicios Ocultos (helper)

Servicio Imagen Cuando corre Descripción
minio-init minio/mc:latest Al iniciar Crea buckets (raw, processed, images)

Guía de Inicio Rápido

# 1. Verificar archivos necesarios
ls -la .env db/data/*.csv db/postgres-sql/init.sql

# 2. Levantar servicios base
docker compose up -d postgres minio

# 3. Verificar salud
docker compose ps

# 4. Ejecutar pipeline
docker compose up workers

# 5. Ver logs en tiempo real
docker compose logs -f workers

Explicación de Logs

Esta es la guía para entender cada línea del output de docker compose logs -f workers:

Formato del Log

[timestamp] [level] [event_name] [logger_name] key=value key=value

Logs del Pipeline (W1 → W2 → W3)

Inicio del Pipeline

2026-04-24T19:53:10.819462Z [info] pipeline.start [__main__] data_source=local env=development file=/app/data/data_api_product_v1.csv version=1.0.0
Campo Valor Significado
timestamp 2026-04-24T19:53:10.819462Z Hora UTC del evento
level info Nivel de log (info/warn/error)
event_name pipeline.start Evento que ocurrió
logger_name __main__ Módulo que generó el log
data_source local Fuente: local (CSV) o s3
env development Ambiente
file /app/data/data_api_product_v1.csv Ruta del CSV
version 1.0.0 Versión del worker

W1: Ingest (Extracción)

# Conexión a base de datos
2026-04-24T19:53:11.068400Z [info] database.engine_created [workers.core.database] db=data-travel-db host=postgres port=5432

# W1 detecta archivo desde variable de entorno
2026-04-24T19:53:11.243170Z [info] w1.using_env_path [workers.pipeline.workers.w1_ingest] path=/app/data/data_api_product_v1.csv

# W1 comienza procesamiento
2026-04-24T19:53:11.243486Z [info] w1.start [workers.pipeline.workers.w1_ingest] file=/app/data/data_api_product_v1.csv size_kb=3.0
Campo Significado
file Ruta del CSV
size_kb Tamaño en KB 
# CSV cargado exitosamente
2026-04-24T19:53:11.274159Z [info] w1.csv_loaded [workers.pipeline.workers.w1_ingest] columns=15 rows=15
Campo Significado
columns Número de columnas en el CSV
rows Número de filas leídas 
# Upload a MinIO bucket raw/
2026-04-24T19:53:11.718833Z [info] s3.upload_ok [workers.storage.s3_minio] bucket=raw key=products/20260424_195311_data_api_product_v1.csv size_bytes=3053
Campo Significado
bucket Bucket de MinIO (raw)
key Ruta del archivo en S3
size_bytes Tamaño del archivo 
# W1 completado
2026-04-24T19:53:11.763520Z [info] w1.done [workers.pipeline.workers.w1_ingest] batch_id=b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf rows_read=15 s3_key=raw/products/20260424_195311_data_api_product_v1.csv
Campo Significado
batch_id UUID único de esta ejecución
rows_read Filas procesadas por W1
s3_key Ruta en MinIO

W2: Clean (Transformación/Validación)

# W2 comienza validación
2026-04-24T19:53:11.764181Z [info] w2.start [workers.pipeline.workers.w2_clean] batch_id=b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf

# W2 completado
2026-04-24T19:53:11.783571Z [info] w2.done [workers.pipeline.workers.w2_clean] batch_id=b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf pct_valid=100.0 rejected=0 total=15 valid=15
Campo Significado
pct_valid Porcentaje de filas válidas
rejected Filas rechazadas por validación
total Total de filas procesadas
valid Filas que pasaron validación

W3: Load (Carga)

# W3 comienza carga
2026-04-24T19:53:11.785169Z [info] w3.start [workers.pipeline.workers.w3_load] batch_id=b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf rows=15

# Insert/Update en PostgreSQL completado
2026-04-24T19:53:11.950424Z [info] w3.db_done [workers.pipeline.workers.w3_load] batch_id=b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf inserted=0 updated=15
Campo Significado
inserted Filas nuevas insertadas
updated Filas actualizadas (upsert) 
# Export a Parquet en MinIO
2026-04-24T19:53:12.076865Z [info] s3.upload_ok [workers.storage.s3_minio] bucket=processed key=products/20260424_195311_b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf.parquet size_bytes=7637

# Parquet exportado
2026-04-24T19:53:12.077704Z [info] w3.parquet_exported [workers.pipeline.workers.w3_load] bucket=processed key=products/20260424_195311_b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf.parquet rows=15 size_bytes=7637

Fin del Pipeline

# Pipeline completado exitosamente
2026-04-24T19:53:12.081844Z [info] pipeline.done [__main__] batch_id=b6a1a65e-3cf0-4e8a-bef7-6e08510c5cbf duration_ms=1259 rows_inserted=0 rows_read=15 rows_rejected=0 rows_updated=15
Campo Significado
batch_id UUID de la ejecución
duration_ms Tiempo total en milisegundos
rows_inserted Total insertados (W3)
rows_read Total leídos (W1)
rows_rejected Total rechazados (W2)
rows_updated Total actualizados (W3)

Errores Comunes y Soluciones

Error 1: "no se puede conectar a Postgres"

Síntoma:

fe_sendauth: no password supplied

Causa: Falta DB_PASSWORD en las variables de entorno del contenedor workers.

Solución: Asegurar que docker-compose.yml tenga todas las variables:

workers:
  environment:
    DB_HOST: postgres
    DB_PORT: 5432
    DB_NAME: ${DB_NAME}
    DB_USER: ${DB_USER}
    DB_PASSWORD: ${DB_PASSWORD}

Error 2: "connection refused" (postgres/minio)

Síntoma:

connection to server at "postgres", port 5432 failed

Causa: Los servicios postgres o minio no están disponibles o no están "healthy".

Solución:

# Verificar estado
docker compose ps

# Reiniciar servicios
docker compose restart postgres minio

Error 3: CSV no se monta en el volumen

Síntoma:

FileNotFoundError: /app/data/data_api_product_v1.csv

Causa: Ruta incorrecta del bind mount o del .env.

Solución:

  1. Verificar que el volumen esté montado: 
    volumes:
  - ./db/data:/app/data
  2. Verificar ruta en .env: 
    LOCAL_DATA_PATH=/app/data/data_api_product_v1.csv

Error 4: "python -m" no ejecuta nada

Síntoma: El contenedor termina instantáneamente sin output.

Causa: Falta el bloque if __name__ == "__main__": en el módulo.

Solución: Agregar al final de orchestrator.py:

if __name__ == "__main__":
    main()

Error 5: Logs no aparecen

Síntoma:

docker compose logs -f workers
# No muestra nada

Causa:

  • El contenedor ya terminó (exit code 0) y logs fueron removidos.
  • El logging va a stderr y no se captura correctamente.

Solución:

# Ver logs del contenedor específico
docker logs data_travel_workers

# O redirigir stderr
docker compose logs -f workers 2>&1

# O ejecutar interactivamente para debug
docker compose run --rm workers

Error 6: MinIO buckets no existen

Síntoma:

NoSuchBucket: The specified bucket does not exist

Causa: El servicio minio-init no se ejecutó o falló.

Solución:

# Recrear buckets manualmente
docker exec data_travel_minio mc mb local/raw
docker exec data_travel_minio mc mb local/processed
docker exec data_travel_minio mc mb local/images

Variables de Entorno

Variables del docker-compose.yml (workers)

Variable Requerido Descripción
DB_HOST ✅ Host de PostgreSQL (docker service name)
DB_PORT ✅ Puerto interno (5432)
DB_NAME ✅ Nombre de la base de datos
DB_USER ✅ Usuario de PostgreSQL
DB_PASSWORD ✅ Contraseña de PostgreSQL
MINIO_ENDPOINT ✅ URL del servicio MinIO
MINIO_ACCESS_KEY ✅ Access key MinIO
MINIO_SECRET_KEY ✅ Secret key MinIO
LOCAL_DATA_PATH ✅ Ruta del CSV dentro del contenedor
LOG_FORMAT Opcional console o json

Comandos Útiles

Levantar servicios

# Todos los servicios
docker compose up -d

# Solo servicios base (sin workers)
docker compose up -d postgres minio

# Reconstruir ylevantar
docker compose up -d --build

Ver estado

# Contenedores corriendo
docker compose ps

# Solo workers
docker compose ps workers

Ver logs

# Todos los logs (follow)
docker compose logs -f

# Solo workers
docker compose logs -f workers

# Últimas 50 líneas
docker compose logs --tail 50 workers

# Logs de contenedor específico
docker logs data_travel_workers

Ejecutar workers manualmente

# Una vez (foreground)
docker compose up workers

# Una vez con rebuild
docker compose up --build workers

# En background
docker compose up -d workers

Debug

# Entrar al contenedor workers
docker compose exec workers /bin/sh

# Probar conexión a PostgreSQL
docker compose exec workers python -c "from workers.core.database import test_check_connection; print(test_check_connection())"

# Probar conexión a MinIO
docker compose exec workers python -c "from workers.storage.s3_minio import get_client; print(get_client().list_buckets())"

Base de datos

# Conectar a PostgreSQL
docker exec -it data_travel_postgres psql -U travel -d data-travel-db

# Ver tabla de productos
docker exec data_travel_postgres psql -U travel -d data-travel-db -c "SELECT * FROM warehouse.products;"

# Ver historial de ejecuciones
docker exec data_travel_postgres psql -U travel -d data-travel-db -c "SELECT * FROM audit.pipeline_runs ORDER BY created_at DESC LIMIT 5;"

# Contar registros
docker exec data_travel_postgres psql -U travel -d data-travel-db -c "SELECT COUNT(*) FROM warehouse.products;"

MinIO (S3)

# Listar buckets
docker exec data_travel_minio mc ls local/

# Listar archivos en raw/
docker exec data_travel_minio mc ls local/raw/

# Listar archivos en processed/
docker exec data_travel_minio mc ls local/processed/

Detener y limpiar

# Detener servicios
docker compose stop

# Detener y eliminar contenedores
docker compose down

# Detener, eliminar contenedores y volúmenes
docker compose down -v

# Eliminar imágenes
docker compose down --rmi all

Diagrama del Flujo

┌────────────────────────────────────────────────────────────────────────────┐
│                 PIPELINE ETL COMPLETO                      │
└────────────────────────────────────────────────────────────┘
        │
        │ CSV File
        ▼
┌────────────────────────────────────────────────────────────┐
│  W1: INGESTA (w1_ingest.py)                           │
│  ├── Lee CSV desde /app/data/                          │
│  ├── Parsea con Polars                               │
│  ├── Sube raw CSV a MinIO (bucket: raw/)             │
│  └── Registra en staging table                      │
└────────────────────────────────────────────────────────────┘
        │
        ▼
┌────────────────────────────────────────────────────────────┐
│  W2: LIMPIEZA (w2_clean.py)                          │
│  ├── Lee de staging table                            │
│  ├── Valida con Pydantic schemas                    │
│  ├── Aplica reglas de negocio                      │
│  └── Separa: valid / rejected                       │
└────────────────────────────────────────────────────────────┘
        │                          │
        ▼                          ▼
┌──────────────────────┐   ┌──────────────────────┐
│     VALID ROWS        │   │   REJECTED ROWS       │
│   (continúan)        │   │   (no se cargan)      │
└──────────────────────┘   └──────────────────────┘
        │
        ▼
┌────────────────────────────────────────────────────────────┐
│  W3: CARGA (w3_load.py)                             │
│  ├── UPSERT a PostgreSQL (catalog.products)         │
│  │   ├── INSERT si no existe                        │
│  │   └── UPDATE si existe (por SKU)                │
│  └── Exporta Parquet a MinIO (bucket: processed/)   │
└────────────────────────────────────────────────────────────┘
        │
        ▼
┌────────────────────────────────────────────────────────────┐
│  REGISTRO EN AUDIT                                   │
│  PipelineRuns table: batch_id, metrics, status       │
└────────────────────────────────────────────────────────────┘
        │
        ▼
      ✓ COMPLETO

Schema de Base de Datos

Tabla: catalog.products

CREATE TABLE catalog.products (
    id SERIAL PRIMARY KEY,
    sku VARCHAR(50) UNIQUE NOT NULL,
    id_producto VARCHAR(50),
    nombre_producto VARCHAR(255),
    descripcion_corta TEXT,
    precio_usd DECIMAL(10,2),
    categoria VARCHAR(100),
    stock_disponible INTEGER,
    fecha_lanzamiento DATE,
    activo BOOLEAN,
    marca VARCHAR(100),
    peso_kg DECIMAL(10,3),
    dimensiones_cm VARCHAR(50),
    valoracion_promedio DECIMAL(3,2),
    num_valoraciones INTEGER,
    url_imagen TEXT,
    batch_id UUID,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

Tabla: audit.pipeline_runs

CREATE TABLE audit.pipeline_runs (
    id UUID PRIMARY KEY,
    source_file VARCHAR(500),
    rows_read INTEGER,
    rows_inserted INTEGER DEFAULT 0,
    rows_update INTEGER DEFAULT 0,
    rows_rejected INTEGER DEFAULT 0,
    status VARCHAR(20),
    worker_version VARCHAR(20),
    error_message TEXT,
    duration_ms INTEGER,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

Notas

  • El pipeline es idempotente: Ejecutar múltiples veces con el mismo CSV actualiza los registros existentes (upsert).
  • Transaccional: Si cualquier worker falla, se hace rollback de toda la operación.
  • Logging estructurado: Los logs usan format JSON para integración con SIEM (Datadog, CloudWatch).
  • Volúmenes Docker: Los datos persisten entre ejecuciones.
  • CSV Encoding: Soporta utf-8-sig (Excel con BOM).

Recursos

Última actualización: 2026-04-24

About

data pipeline platform

Resources

Readme
Activity

Stars

1 star

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages