Cómo crear un pipeline ETL en Python con pandas

Mover datos de un archivo en bruto a una base de datos lista para consultar es una de las tareas más comunes en cualquier proyecto de datos. Un pipeline ETL (Extract, Transform, Load) hace exactamente eso de forma organizada y repetible. Aquí vas a construir, de principio a fin, un pipeline ETL sencillo en Python que lee un CSV de ventas, limpia y enriquece los datos, y carga el resultado en una base de datos SQLite.

Requisitos previos

• Python 3.9 o superior instalado.
• La biblioteca pandas (se instala con pip install pandas).
• Conocimientos básicos de Python (variables y funciones).
• Un editor de código o una terminal. SQLite ya viene incluido en Python, a través del módulo sqlite3.

Paso 1: Preparar el entorno y los datos

Empieza por instalar pandas y crear un archivo de ejemplo. Este pequeño script genera un vendas.csv con algunas filas, incluyendo un valor ausente a propósito, para ver cómo tratarlo más adelante.

pip install pandas

import pandas as pd

dados = {
    "data": ["2026-06-01", "2026-06-01", "2026-06-02", None],
    "produto": ["Teclado", "Rato", "Monitor", "Teclado"],
    "quantidade": [3, 5, 2, 1],
    "preco_unitario": [25.0, 12.5, 150.0, 25.0],
}
pd.DataFrame(dados).to_csv("vendas.csv", index=False)

Paso 2: Extract — leer los datos de origen

La fase de extracción lee los datos de la fuente. Con pandas, leer un CSV es una sola línea. Imprime las primeras filas para confirmar que los datos llegaron como esperabas antes de transformarlos.

import pandas as pd

df = pd.read_csv("vendas.csv")
print(df.head())

Paso 3: Transform — limpiar y enriquecer

La transformación es el corazón del pipeline. Aquí eliminamos filas inválidas, convertimos las columnas a los tipos de datos correctos y creamos una columna nueva con el valor total de cada venta. Este es también el lugar donde, en un proyecto real, aplicarías tus reglas de negocio.

# Remover linhas sem data
df = df.dropna(subset=["data"])

# Converter para os tipos corretos
df["data"] = pd.to_datetime(df["data"])
df["quantidade"] = df["quantidade"].astype(int)

# Criar uma coluna derivada
df["total"] = df["quantidade"] * df["preco_unitario"]

Paso 4: Load — cargar en la base de datos

La fase de carga escribe los datos ya tratados en el destino. El método to_sql de pandas crea la tabla e inserta las filas en una base de datos SQLite. El parámetro if_exists="replace" vuelve a crear la tabla en cada ejecución, lo cual es práctico mientras haces pruebas.

import sqlite3

conn = sqlite3.connect("vendas.db")
df.to_sql("vendas", conn, if_exists="replace", index=False)
conn.commit()
conn.close()

Paso 5: Unir todo en un único pipeline

Con las tres fases listas, el paso final es organizarlas en una función reutilizable. Así puedes ejecutar todo el proceso con una sola llamada y reutilizarlo para otros archivos.

def etl(origem_csv, destino_db):
    df = pd.read_csv(origem_csv)
    df = df.dropna(subset=["data"])
    df["data"] = pd.to_datetime(df["data"])
    df["total"] = df["quantidade"] * df["preco_unitario"]

    conn = sqlite3.connect(destino_db)
    df.to_sql("vendas", conn, if_exists="replace", index=False)
    conn.close()
    return len(df)

linhas = etl("vendas.csv", "vendas.db")
print(f"Carregadas {linhas} linhas")

Verificar el resultado

Para confirmar que todo salió bien, vuelve a conectarte a la base de datos y consulta la tabla. Deberías ver tres filas (la fila sin fecha se eliminó) y la nueva columna total rellena.

import sqlite3
import pandas as pd

conn = sqlite3.connect("vendas.db")
print(pd.read_sql("SELECT * FROM vendas", conn))
conn.close()

Si la consulta devuelve las filas con las columnas data, produto, quantidade, preco_unitario y total, tu pipeline ETL está funcionando.

Conclusión

Has construido un pipeline ETL completo: extrajiste datos de un CSV, los transformaste con pandas y los cargaste en una base de datos SQLite lista para consultar. A partir de aquí, los próximos pasos naturales son programar el script (con cron en Linux o el Task Scheduler en Windows), añadir cargas incrementales para procesar solo filas nuevas y, a medida que el volumen crece, migrar a una herramienta de orquestación como Azure Data Factory. ¿Qué fuente de datos vas a transformar a continuación en tu primer pipeline?