Formato CSV: intercambio de datos tabulares con Python

Qué es CSV y por qué es el rey del intercambio de datos

CSV (Comma-Separated Values) es el formato de intercambio de datos tabulares más utilizado en el mundo. A pesar de su simplicidad — o precisamente por ella — sigue siendo el estándar de facto para:

• Bases de datos: exportar e importar tablas en MySQL, PostgreSQL, SQLite
• Hojas de cálculo: Excel, Google Sheets leen y escriben CSV nativamente
• APIs y servicios: muchas APIs ofrecen resultados en CSV además de JSON
• Machine Learning: scikit-learn, XGBoost y PyTorch aceptan CSV como entrada
• ETL: casi todos los pipelines de datos incluyen pasos con CSV

Un archivo CSV es texto plano: una fila por registro, campos separados por comas (o punto y coma, tabulaciones, etc.), con una línea de cabecera opcional.

Variantes comunes

Módulo csv de la biblioteca estándar

Para archivos medianos sin dependencias externas:

import csv
from pathlib import Path

# ── Lectura básica ─────────────────────────────────────────────────────────
def leer_csv(ruta, delimitador=',', encoding='utf-8'):
    """Lee un CSV y devuelve lista de diccionarios."""
    with open(ruta, newline='', encoding=encoding) as f:
        lector = csv.DictReader(f, delimiter=delimitador)
        filas  = list(lector)
    print(f"Leídas {len(filas)} filas de {ruta}")
    return filas


# ── Escritura básica ───────────────────────────────────────────────────────
def escribir_csv(datos, ruta, campos=None, delimitador=',', encoding='utf-8'):
    """
    Escribe una lista de diccionarios a CSV.
    Si campos=None, usa las claves del primer elemento.
    """
    if not datos:
        print("Lista vacía — nada que escribir")
        return

    campos = campos or list(datos[0].keys())

    with open(ruta, 'w', newline='', encoding=encoding) as f:
        escritor = csv.DictWriter(f, fieldnames=campos,
                                   delimiter=delimitador,
                                   extrasaction='ignore')
        escritor.writeheader()
        escritor.writerows(datos)

    tamaño = Path(ruta).stat().st_size / 1024
    print(f"CSV escrito: {ruta} ({len(datos)} filas, {tamaño:.1f} KB)")


# ── Detectar dialecto automáticamente ─────────────────────────────────────
def detectar_dialecto(ruta, muestra_bytes=4096):
    """Detecta el separador y dialecto de un CSV desconocido."""
    with open(ruta, newline='', encoding='utf-8-sig') as f:
        muestra = f.read(muestra_bytes)

    dialecto = csv.Sniffer().sniff(muestra, delimiters=',;\t|')
    tiene_cabecera = csv.Sniffer().has_header(muestra)

    print(f"Separador detectado: '{dialecto.delimiter}'")
    print(f"Tiene cabecera: {tiene_cabecera}")
    print(f"Comillas: '{dialecto.quotechar}'")
    return dialecto


# Ejemplo de uso
empleados = [
    {'nombre': 'Ana García',    'puesto': 'Ingeniería',  'salario': 55000},
    {'nombre': 'Luis Martínez', 'puesto': 'Marketing',   'salario': 48000},
    {'nombre': 'Sara López',    'puesto': 'Diseño',      'salario': 52000},
]
escribir_csv(empleados, 'empleados.csv')
leídos = leer_csv('empleados.csv')

CSV con pandas (análisis y transformaciones)

import pandas as pd
import numpy as np
from pathlib import Path

def leer_csv_pandas(ruta, **kwargs):
    """
    Lee CSV con inferencia automática de tipos.
    Usa utf-8-sig para manejar BOM de Excel.
    """
    defaults = {
        'encoding':        'utf-8-sig',
        'on_bad_lines':    'warn',
        'dtype_backend':   'numpy_nullable',  # mejor manejo de nulos
    }
    defaults.update(kwargs)

    df = pd.read_csv(ruta, **defaults)
    print(f"DataFrame: {df.shape[0]:,} filas × {df.shape[1]} columnas")
    print(f"Tipos:\n{df.dtypes.to_string()}")
    print(f"Nulos:\n{df.isnull().sum().to_string()}")
    return df


def limpiar_csv(ruta_entrada, ruta_salida):
    """Pipeline de limpieza de datos CSV."""
    df = pd.read_csv(ruta_entrada, encoding='utf-8-sig')

    print(f"Antes: {df.shape}")

    # 1. Eliminar filas completamente vacías
    df = df.dropna(how='all')

    # 2. Normalizar nombres de columnas (snake_case, sin acentos)
    import unicodedata, re
    def normalizar_col(nombre):
        nombre = unicodedata.normalize('NFD', str(nombre))
        nombre = ''.join(c for c in nombre if unicodedata.category(c) != 'Mn')
        nombre = nombre.lower().strip()
        nombre = re.sub(r'[^\w\s]', '', nombre)
        nombre = re.sub(r'\s+', '_', nombre)
        return nombre

    df.columns = [normalizar_col(c) for c in df.columns]

    # 3. Eliminar duplicados
    n_antes = len(df)
    df = df.drop_duplicates()
    n_dup = n_antes - len(df)
    if n_dup:
        print(f"  Duplicados eliminados: {n_dup}")

    # 4. Limpiar strings: quitar espacios extra
    str_cols = df.select_dtypes(include='object').columns
    for col in str_cols:
        df[col] = df[col].str.strip()

    # 5. Inferir tipos numéricos donde sea posible
    df = df.infer_objects()
    for col in df.select_dtypes(include='object').columns:
        try:
            df[col] = pd.to_numeric(df[col].str.replace(',', '.'), errors='ignore')
        except Exception:
            pass

    print(f"Después: {df.shape}")
    df.to_csv(ruta_salida, index=False, encoding='utf-8')
    print(f"CSV limpio guardado: {ruta_salida}")
    return df


def analizar_csv(ruta):
    """Análisis rápido de calidad de datos."""
    df = pd.read_csv(ruta, encoding='utf-8-sig')

    print(f"\n{'='*50}")
    print(f"ANÁLISIS: {Path(ruta).name}")
    print(f"{'='*50}")
    print(f"Dimensiones: {df.shape[0]:,} filas × {df.shape[1]} columnas")
    print(f"\nColumnas y tipos:")

    for col in df.columns:
        nulos = df[col].isnull().sum()
        pct_nulos = nulos / len(df) * 100
        nunique = df[col].nunique()
        print(f"  {col:20s} | {str(df[col].dtype):10s} | "
              f"nulos: {pct_nulos:5.1f}% | únicos: {nunique:,}")

    print(f"\nEstadísticas numéricas:")
    num_cols = df.select_dtypes(include=np.number)
    if not num_cols.empty:
        print(num_cols.describe().round(2).to_string())


# Ejemplo
df = pd.DataFrame({
    'nombre':  ['Ana García', 'Luis Martínez', 'Sara López', 'Ana García'],
    'edad':    [28, 35, 31, 28],
    'salario': ['55.000', '48.000', '52.000', '55.000'],
    'ciudad':  ['Madrid', 'Barcelona', 'Sevilla', 'Madrid'],
})
df.to_csv('datos_sucios.csv', index=False)
limpiar_csv('datos_sucios.csv', 'datos_limpios.csv')

Convertir CSV a otros formatos

def csv_a_json(csv_ruta, json_ruta, orientacion='records', indent=2):
    """
    Convierte CSV a JSON.
    orientacion: 'records' (lista), 'split', 'index', 'columns', 'values'
    """
    df = pd.read_csv(csv_ruta, encoding='utf-8-sig')
    df.to_json(json_ruta, orient=orientacion, indent=indent,
               force_ascii=False)
    tamaño = Path(json_ruta).stat().st_size / 1024
    print(f"CSV→JSON: {json_ruta} ({tamaño:.1f} KB)")


def csv_a_excel(csv_ruta, excel_ruta, nombre_hoja='Datos'):
    """Convierte CSV a XLSX con formato básico."""
    df = pd.read_csv(csv_ruta, encoding='utf-8-sig')

    with pd.ExcelWriter(excel_ruta, engine='openpyxl') as writer:
        df.to_excel(writer, sheet_name=nombre_hoja, index=False)

        # Ajustar ancho de columnas
        hoja = writer.sheets[nombre_hoja]
        for col_idx, col in enumerate(df.columns, 1):
            max_len = max(
                df[col].astype(str).map(len).max(),
                len(col)
            ) + 2
            from openpyxl.utils import get_column_letter
            hoja.column_dimensions[get_column_letter(col_idx)].width = min(max_len, 40)

    print(f"CSV→Excel: {excel_ruta}")


def csv_a_parquet(csv_ruta, parquet_ruta, compresion='snappy'):
    """
    Convierte CSV a Parquet para análisis de big data.
    Parquet es columnar, comprimido y mucho más rápido que CSV para consultas.
    """
    df = pd.read_csv(csv_ruta, encoding='utf-8-sig')
    df.to_parquet(parquet_ruta, compression=compresion, index=False)

    tamaño_csv = Path(csv_ruta).stat().st_size / 1024
    tamaño_par = Path(parquet_ruta).stat().st_size / 1024
    reduccion  = (1 - tamaño_par / tamaño_csv) * 100
    print(f"CSV→Parquet: {parquet_ruta} "
          f"({tamaño_csv:.0f} KB → {tamaño_par:.0f} KB, -{reduccion:.0f}%)")


def csv_a_sqlite(csv_ruta, db_ruta, nombre_tabla=None):
    """Importa un CSV directamente a una tabla SQLite."""
    import sqlite3
    df = pd.read_csv(csv_ruta, encoding='utf-8-sig')

    if nombre_tabla is None:
        nombre_tabla = Path(csv_ruta).stem.lower().replace(' ', '_')

    with sqlite3.connect(db_ruta) as conn:
        df.to_sql(nombre_tabla, conn, if_exists='replace', index=False)

    filas = len(df)
    print(f"CSV→SQLite: tabla '{nombre_tabla}' ({filas:,} filas) en {db_ruta}")


# Conversiones
csv_a_json('datos.csv', 'datos.json')
csv_a_excel('datos.csv', 'datos.xlsx')
csv_a_parquet('datos.csv', 'datos.parquet')
csv_a_sqlite('datos.csv', 'base.db')

Procesar CSV grandes sin cargar en memoria

def procesar_csv_grande(ruta, funcion_filtro, ruta_salida, chunk_size=50_000):
    """
    Procesa CSVs grandes por chunks sin cargar todo en memoria.
    Ideal para archivos de varios GB.
    """
    total_procesadas = 0
    total_filtradas  = 0
    primera_escritura = True

    for chunk in pd.read_csv(ruta, chunksize=chunk_size, encoding='utf-8-sig'):
        total_procesadas += len(chunk)

        # Aplicar filtro personalizable
        chunk_filtrado = funcion_filtro(chunk)
        total_filtradas += len(chunk_filtrado)

        if not chunk_filtrado.empty:
            chunk_filtrado.to_csv(
                ruta_salida,
                mode='w' if primera_escritura else 'a',
                index=False,
                header=primera_escritura
            )
            primera_escritura = False

        print(f"\rProcesadas: {total_procesadas:,} filas...", end='')

    print(f"\nTotal: {total_procesadas:,} procesadas → {total_filtradas:,} filtradas")
    print(f"Resultado: {ruta_salida}")


def combinar_csvs(patron_o_lista, salida, deduplicar=True):
    """Combina múltiples CSVs en uno solo."""
    if isinstance(patron_o_lista, str):
        archivos = sorted(Path('.').glob(patron_o_lista))
    else:
        archivos = [Path(f) for f in patron_o_lista]

    dfs = []
    for archivo in archivos:
        df = pd.read_csv(archivo, encoding='utf-8-sig')
        df['_origen'] = archivo.name  # columna de trazabilidad
        dfs.append(df)
        print(f"  Cargado: {archivo.name} ({len(df):,} filas)")

    combinado = pd.concat(dfs, ignore_index=True)

    if deduplicar:
        n_antes = len(combinado)
        combinado = combinado.drop_duplicates(
            subset=[c for c in combinado.columns if c != '_origen']
        )
        print(f"Duplicados eliminados: {n_antes - len(combinado):,}")

    combinado.to_csv(salida, index=False, encoding='utf-8')
    print(f"Combinado: {len(archivos)} archivos → {salida} ({len(combinado):,} filas)")
    return combinado


# Ejemplo: filtrar CSV grande
def filtro_activos(chunk):
    """Ejemplo: solo registros activos con salario > 30000."""
    return chunk[(chunk.get('activo', True) == True) &
                 (chunk.get('salario', 0) > 30000)]

procesar_csv_grande('empleados_grandes.csv', filtro_activos,
                     'empleados_activos.csv')

Validación de datos CSV

def validar_csv(ruta, esquema):
    """
    Valida un CSV contra un esquema de reglas.
    esquema = {
        'columna': {'tipo': 'int', 'min': 0, 'max': 100, 'requerida': True}
    }
    """
    df = pd.read_csv(ruta, encoding='utf-8-sig')
    errores = []

    for columna, reglas in esquema.items():
        if reglas.get('requerida') and columna not in df.columns:
            errores.append(f"Columna requerida ausente: '{columna}'")
            continue

        if columna not in df.columns:
            continue

        col = df[columna]

        # Verificar nulos
        n_nulos = col.isnull().sum()
        if reglas.get('requerida') and n_nulos > 0:
            errores.append(f"'{columna}': {n_nulos} valores nulos no permitidos")

        # Verificar tipo
        tipo_esperado = reglas.get('tipo')
        if tipo_esperado == 'int':
            no_int = col.dropna()[~col.dropna().apply(
                lambda x: str(x).lstrip('-').isdigit()
            )]
            if not no_int.empty:
                errores.append(f"'{columna}': {len(no_int)} valores no son enteros")

        # Verificar rango
        if reglas.get('min') is not None:
            bajo = (pd.to_numeric(col, errors='coerce') < reglas['min']).sum()
            if bajo:
                errores.append(f"'{columna}': {bajo} valores < {reglas['min']}")

        if reglas.get('max') is not None:
            alto = (pd.to_numeric(col, errors='coerce') > reglas['max']).sum()
            if alto:
                errores.append(f"'{columna}': {alto} valores > {reglas['max']}")

    if errores:
        print(f"❌ Validación fallida ({len(errores)} errores):")
        for e in errores:
            print(f"  - {e}")
    else:
        print(f"✅ CSV válido: {ruta}")

    return errores


# Esquema de ejemplo
esquema_empleados = {
    'nombre':  {'requerida': True},
    'edad':    {'requerida': True, 'tipo': 'int', 'min': 18, 'max': 99},
    'salario': {'requerida': True, 'min': 0, 'max': 500_000},
}
validar_csv('empleados.csv', esquema_empleados)

Conclusión

CSV es el mínimo común denominador del intercambio de datos: legible por humanos, soportado universalmente y generado por cualquier herramienta. Python ofrece tres capas para trabajar con él:

1. Módulo csv — para archivos pequeños sin dependencias
2. pandas — para análisis, limpieza y transformaciones (hasta ~2 GB en RAM)
3. Polars o Dask — para archivos de decenas de GB sin cargar en memoria

La conversión CSV → Parquet es una de las transformaciones más valiosas en data engineering: mismos datos pero 5-20× más rápido en consultas y 60-80% menos espacio en disco.

Conversiones relacionadas

Conversiones de documento que siguen este tema:

• PDF a DOCX
• DOCX a PDF
• MD a PDF
• HTML a PDF
• ODT a PDF
• PDF a TXT