Proyecto: entorno de desarrollo completo con docker-compose

1# Estructura de archivos del proyecto
2mi-pipeline-docker/
3├── docker-compose.yml      # Orquesta todos los servicios
4├── Dockerfile              # Imagen personalizada para el ETL
5├── requirements.txt        # Dependencias Python del ETL
6├── etl/
7│   └── pipeline.py         # El script ETL principal
8├── data/
9│   └── ventas_raw.csv      # Datos crudos de entrada
10└── init-db/
11    └── schema.sql          # Esquema de la tabla destino

Estructura profesional del proyecto — limpia y organizada

1# Archivo: data/ventas_raw.csv
2id,producto,cantidad,precio_unitario,fecha,cliente,ciudad
31,Laptop Pro,2,1299.99,2025-01-15,Ana García,Madrid
42,Mouse Wireless,,29.99,2025-01-15,Carlos López,Barcelona
53,Teclado Mecánico,1,89.50,15/01/2025,María Fernández,Valencia
64,Monitor 4K,1,549.00,2025-01-16,,Sevilla
75,Laptop Pro,3,1299.99,2025-01-16,Ana García,Madrid
86,USB Hub,-1,19.99,2025-01-17,Pedro Martín,Bilbao
97,Webcam HD,2,79.99,2025-01-17,Laura Sánchez,Madrid
108,Monitor 4K,1,549.00,2025-01-18,Carlos López,Barcelona

Datos crudos con errores típicos: campos vacíos, fechas inconsistentes, cantidad negativa

1-- Archivo: init-db/schema.sql
2CREATE TABLE IF NOT EXISTS ventas_limpias (
3    id SERIAL PRIMARY KEY,
4    producto VARCHAR(100) NOT NULL,
5    cantidad INTEGER NOT NULL CHECK (cantidad > 0),
6    precio_unitario DECIMAL(10,2) NOT NULL,
7    total DECIMAL(10,2) NOT NULL,
8    fecha DATE NOT NULL,
9    cliente VARCHAR(100),
10    ciudad VARCHAR(50) NOT NULL,
11    cargado_en TIMESTAMP DEFAULT NOW()
12);

schema.sql — la tabla destino con validaciones

1# Archivo: etl/pipeline.py
2"""Mini pipeline ETL: CSV → limpieza → PostgreSQL"""
3import csv
4import psycopg2
5from datetime import datetime
6import time
7import os
8
9def esperar_postgres(max_intentos=10):
10    """Espera a que PostgreSQL esté listo antes de conectar."""
11    for intento in range(max_intentos):
12        try:
13            conn = psycopg2.connect(
14                host=os.getenv("DB_HOST", "postgres"),
15                port=5432,
16                user=os.getenv("DB_USER", "etl_user"),
17                password=os.getenv("DB_PASS", "etl_secreto"),
18                database=os.getenv("DB_NAME", "ventas")
19            )
20            conn.close()
21            print(f"✅ PostgreSQL listo (intento {intento + 1})")
22            return True
23        except psycopg2.OperationalError:
24            print(f"⏳ Esperando PostgreSQL... (intento {intento + 1})")
25            time.sleep(2)
26    raise Exception("❌ PostgreSQL no respondió después de varios intentos")
27
28def leer_csv(filepath):
29    """Lee el CSV crudo."""
30    with open(filepath, 'r', encoding='utf-8') as f:
31        reader = csv.DictReader(f)
32        return list(reader)
33
34def limpiar_fila(fila):
35    """Limpia y valida una fila. Retorna None si no es válida."""
36    # Validar cantidad
37    try:
38        cantidad = int(fila['cantidad'])
39        if cantidad <= 0:
40            print(f"  ⚠️  Fila {fila['id']}: cantidad inválida ({cantidad})")
41            return None
42    except (ValueError, TypeError):
43        print(f"  ⚠️  Fila {fila['id']}: cantidad vacía o no numérica")
44        return None
45
46    # Validar cliente
47    cliente = fila.get('cliente', '').strip() or 'Desconocido'
48
49    # Normalizar fecha (soportar YYYY-MM-DD y DD/MM/YYYY)
50    fecha_raw = fila['fecha'].strip()
51    try:
52        if '/' in fecha_raw:
53            fecha = datetime.strptime(fecha_raw, '%d/%m/%Y').date()
54        else:
55            fecha = datetime.strptime(fecha_raw, '%Y-%m-%d').date()
56    except ValueError:
57        print(f"  ⚠️  Fila {fila['id']}: fecha inválida ({fecha_raw})")
58        return None
59
60    # Validar precio
61    try:
62        precio = float(fila['precio_unitario'])
63        if precio <= 0:
64            print(f"  ⚠️  Fila {fila['id']}: precio inválido ({precio})")
65            return None
66    except (ValueError, TypeError):
67        print(f"  ⚠️  Fila {fila['id']}: precio vacío o no numérico")
68        return None
69
70    total = cantidad * precio
71
72    return {
73        'producto': fila['producto'].strip(),
74        'cantidad': cantidad,
75        'precio_unitario': precio,
76        'total': round(total, 2),
77        'fecha': fecha,
78        'cliente': cliente,
79        'ciudad': fila['ciudad'].strip(),
80    }
81
82def cargar_en_postgres(filas_limpias):
83    """Inserta las filas limpias en PostgreSQL."""
84    conn = psycopg2.connect(
85        host=os.getenv("DB_HOST", "postgres"),
86        port=5432,
87        user=os.getenv("DB_USER", "etl_user"),
88        password=os.getenv("DB_PASS", "etl_secreto"),
89        database=os.getenv("DB_NAME", "ventas")
90    )
91    cursor = conn.cursor()
92
93    for fila in filas_limpias:
94        cursor.execute("""
95            INSERT INTO ventas_limpias
96            (producto, cantidad, precio_unitario, total, fecha, cliente, ciudad)
97            VALUES (%s, %s, %s, %s, %s, %s, %s)
98        """, (
99            fila['producto'], fila['cantidad'], fila['precio_unitario'],
100            fila['total'], fila['fecha'], fila['cliente'], fila['ciudad']
101        ))
102
103    conn.commit()
104    cursor.close()
105    conn.close()
106    return len(filas_limpias)
107
108if __name__ == '__main__':
109    print("=" * 50)
110    print("🚀 Pipeline ETL iniciado")
111    print("=" * 50)
112
113    esperar_postgres()
114
115    # EXTRACT
116    print("\n📥 Extrayendo datos del CSV...")
117    datos_crudos = leer_csv('/app/data/ventas_raw.csv')
118    print(f"   Filas leídas: {len(datos_crudos)}")
119
120    # TRANSFORM
121    print("\n🔧 Limpiando y transformando...")
122    datos_limpios = []
123    for fila in datos_crudos:
124        limpia = limpiar_fila(fila)
125        if limpia:
126            datos_limpios.append(limpia)
127
128    print(f"   Filas válidas: {len(datos_limpios)}/{len(datos_crudos)}")
129
130    # LOAD
131    print("\n📤 Cargando en PostgreSQL...")
132    insertadas = cargar_en_postgres(datos_limpios)
133    print(f"   Filas insertadas: {insertadas}")
134
135    print("\n✅ Pipeline completado exitosamente!")
136    print("=" * 50)

pipeline.py — ETL completo: Extract, Transform, Load

1# Archivo: Dockerfile
2FROM python:3.12-slim
3
4WORKDIR /app
5
6COPY requirements.txt .
7RUN pip install --no-cache-dir -r requirements.txt
8
9COPY etl/ ./etl/
10COPY data/ ./data/
11
12CMD ["python", "etl/pipeline.py"]

Dockerfile del ETL — simple y efectivo

1# Archivo: requirements.txt
2psycopg2-binary==2.9.9

requirements.txt — solo necesitamos el driver de PostgreSQL

1# Archivo: docker-compose.yml
2services:
3  postgres:
4    image: postgres:16
5    environment:
6      POSTGRES_USER: etl_user
7      POSTGRES_PASSWORD: etl_secreto
8      POSTGRES_DB: ventas
9    ports:
10      - "5432:5432"
11    volumes:
12      - pgdata:/var/lib/postgresql/data
13      - ./init-db:/docker-entrypoint-initdb.d  # Ejecuta schema.sql al inicio
14
15  pgadmin:
16    image: dpage/pgadmin4:latest
17    environment:
18      PGADMIN_DEFAULT_EMAIL: admin@datos.com
19      PGADMIN_DEFAULT_PASSWORD: admin123
20    ports:
21      - "8080:80"
22    depends_on:
23      - postgres
24
25  etl:
26    build: .
27    environment:
28      DB_HOST: postgres
29      DB_USER: etl_user
30      DB_PASS: etl_secreto
31      DB_NAME: ventas
32    depends_on:
33      - postgres
34    volumes:
35      - ./data:/app/data  # Montar los CSVs
36
37volumes:
38  pgdata:

docker-compose.yml final — 3 servicios orquestados

1# Levantar todo (construye la imagen ETL + arranca PostgreSQL + pgAdmin)
2docker compose up --build
3
4# Verás los logs de los 3 servicios:
5# postgres  | PostgreSQL init process complete
6# pgadmin   | Listening at: http://0.0.0.0:80
7# etl       | 🚀 Pipeline ETL iniciado
8# etl       | ✅ PostgreSQL listo
9# etl       | 📥 Extrayendo datos del CSV...
10# etl       | 🔧 Limpiando y transformando...
11# etl       | 📤 Cargando en PostgreSQL...
12# etl       | ✅ Pipeline completado exitosamente!
13
14# El ETL termina pero PostgreSQL y pgAdmin siguen corriendo
15# Abre http://localhost:8080 para ver los datos en pgAdmin

Un comando y todo funciona — la magia de docker-compose

1# Conectarte directamente a PostgreSQL desde la terminal
2docker compose exec postgres psql -U etl_user -d ventas
3
4# Dentro de psql:
5# ventas=# SELECT * FROM ventas_limpias;
6# ventas=# SELECT COUNT(*) FROM ventas_limpias;
7# ventas=# SELECT ciudad, SUM(total) FROM ventas_limpias GROUP BY ciudad;
8# ventas=# \q  (para salir)

Verificar que los datos llegaron correctamente a PostgreSQL

1# 1. Los tres servicios arriba, postgres marcado (healthy)
2docker compose ps --format "{{.Service}}  {{.Status}}"
3
4# 2. Tu ETL cargó datos en la base
5docker compose exec postgres psql -U etl_user -d ventas \
6  -tAc "SELECT COUNT(*) FROM ventas_limpias;"
7
8# 3. El reporte existe
9cat output/reporte.json
10
11# 4. Tu imagen es pequeña (multi-stage)
12docker images --format "{{.Repository}}:{{.Tag}}  {{.Size}}" | grep etl
13
14# 5. El compose sigue siendo válido
15docker compose config --services | sort

Si los cinco comandos dan resultado, tu entorno profesional está completo.