StackPractices
beginner Por Mathias Paulenko

Importar Datos desde CSV/Excel

Cómo parsear e importar datos desde archivos CSV y Excel con validación, streaming e inserción por lotes.

Visión general

Importar datos desde CSV o Excel es una funcionalidad estándar de paneles de administración, herramientas de migración y flujos de actualización masiva. La parte difícil rara vez es el parseo: es todo lo que lo rodea. Los usuarios suben archivos con el encoding equivocado, columnas renombradas, filas en blanco y fechas guardadas como números de serie en punto flotante. Una hoja de cálculo de 500MB se comerá toda tu RAM encantada si la cargas sin pensar.

Esta receta cubre la importación confiable de CSV/Excel en Python, JavaScript y Java: parsers de streaming, validación fila a fila con colección de errores, detección de encoding e inserción por lotes. Si solo necesitas leer un archivo — sin ingestarlo en un sistema — mira Parsear archivos CSV o Parsear archivos Excel, que cubren el camino más ligero de solo lectura.

Cuándo usarla

Usa este recurso cuando:

  • Los usuarios suban hojas de cálculo para crear o actualizar registros masivamente. Consulta Validación de subida de archivos para asegurar la subida en sí, no solo los datos.
  • Estés migrando datos desde sistemas legacy o proveedores externos. Consulta Exportar a CSV/Excel para la dirección inversa.
  • Estés construyendo pipelines ETL que procesen entregas de archivos programadas.
  • Un panel de administración necesite una capacidad de “subir e importar”. Consulta Trabajos en segundo plano para sacar las importaciones largas del hilo de la petición.

Solución

Python (pandas + csv)

import csv
import pandas as pd
from pydantic import BaseModel, ValidationError

# Importación CSV en streaming con validación fila a fila
class UserImport(BaseModel):
    name: str
    email: str
    age: int

def import_users_csv(file_path):
    valid, errors = [], []
    with open(file_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row_num, row in enumerate(reader, start=2):
            try:
                user = UserImport(**row)
                valid.append(user.model_dump())  # Pydantic v2; usa .dict() en v1
            except (ValidationError, ValueError) as e:
                errors.append({"row": row_num, "error": str(e)})
    return valid, errors

# Importación Excel con pandas
def import_users_excel(file_path):
    df = pd.read_excel(file_path, sheet_name="Users")
    df = df.dropna(how="all")  # descarta filas completamente vacías, conserva las parciales
    return df.to_dict("records")

JavaScript (csv-parser + xlsx)

const csv = require("csv-parser");
const fs = require("fs");
const XLSX = require("xlsx");

function importCsv(filePath) {
  return new Promise((resolve, reject) => {
    const valid = [], errors = [];
    let rowNum = 1;

    fs.createReadStream(filePath)
      .pipe(csv())
      .on("data", (row) => {
        rowNum++;
        if (!row.name || !row.email) {
          errors.push({ row: rowNum, error: "Missing required field" });
        } else {
          valid.push(row);
        }
      })
      .on("end", () => resolve({ valid, errors }))
      .on("error", reject);
  });
}

function importExcel(filePath) {
  const workbook = XLSX.readFile(filePath);
  const sheet = workbook.Sheets[workbook.SheetNames[0]];
  const rows = XLSX.utils.sheet_to_json(sheet, { header: 1 });
  const headers = rows[0];
  return rows.slice(1).map((row) => {
    const obj = {};
    headers.forEach((h, i) => (obj[h] = row[i]));
    return obj;
  });
}

Java (Apache Commons CSV + Apache POI)

import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.apache.poi.ss.usermodel.*;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import java.io.*;
import java.nio.file.*;
import java.util.*;

public class Importer {

    public List<Map<String, String>> importCsv(Path path) throws IOException {
        List<Map<String, String>> rows = new ArrayList<>();
        try (Reader reader = Files.newBufferedReader(path);
             CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT.withFirstRecordAsHeader())) {
            for (CSVRecord record : parser) {
                Map<String, String> row = new HashMap<>();
                record.toMap().forEach(row::put);
                rows.add(row);
            }
        }
        return rows;
    }

    public List<Map<String, String>> importExcel(Path path) throws IOException {
        List<Map<String, String>> rows = new ArrayList<>();
        try (Workbook workbook = new XSSFWorkbook(Files.newInputStream(path))) {
            Sheet sheet = workbook.getSheetAt(0);
            Row headerRow = sheet.getRow(0);
            List<String> headers = new ArrayList<>();
            for (Cell cell : headerRow) {
                headers.add(cell.getStringCellValue());
            }
            for (int i = 1; i <= sheet.getLastRowNum(); i++) {
                Row row = sheet.getRow(i);
                if (row == null) continue;
                Map<String, String> map = new HashMap<>();
                for (int j = 0; j < headers.size(); j++) {
                    Cell cell = row.getCell(j);
                    map.put(headers.get(j), cell != null ? cell.toString() : "");
                }
                rows.add(map);
            }
        }
        return rows;
    }
}

Explicación

Importar es más difícil que exportar porque no controlas la entrada. Los usuarios subirán archivos con headers faltantes, tipos incorrectos, filas en blanco extra y el encoding que a su Excel le pareciera. El patrón confiable es un pipeline:

  1. Parsear el archivo fila por fila. El streaming importa — csv.DictReader, csv-parser y CSVParser iteran sin cargar el archivo entero en memoria.
  2. Validar cada fila contra un esquema. En Python, deja que Pydantic haga la coerción y la validación en vez de convertir int(row["age"]) a mano — obtienes errores más claros gratis.
  3. Recolectar errores en lugar de parar en la primera fila mala. Quien corrige un CSV quiere la lista completa de errores de una vez, con números de fila.
  4. Insertar por lotes: para importaciones a base de datos, inserta en bloques dentro de una transacción (ver Importar a una base de datos más abajo).

Dos reglas prácticas: nunca confíes en el archivo y nunca obligues al usuario a reintentar. Los archivos que pasan la validación siguen escondiendo sorpresas — una columna llamada email (espacio al final), una hoja donde la fila 1 es un título decorativo en vez de headers — así que normaliza los headers (.strip().lower()) antes de casarlos con tu esquema.

Variantes

FormatoLibrería¿Streaming?Ideal para
CSVPython csvSíArchivos pequeños a medianos con validación propia
CSVcsv-parser (JS)SíPipeline de streaming en Node.js
CSVApache Commons CSVSíParseo Java enterprise
Excelpandas.read_excelNoImportaciones rápidas, exploración de datos
Excelxlsx (JS)NoCliente o importaciones pequeñas en servidor
ExcelApache POISí (event model)Archivos Excel muy grandes

El event model de Apache POI (XSSFReader + SAX) es el único de esta lista que hace streaming de Excel — vale la pena cuando un cliente te entrega un workbook de 200.000 filas. pandas.read_excel carga la hoja entera en un DataFrame, que va bien hasta unos cientos de megabytes pero es una pared más allá.

Archivos del mundo real

Las soluciones de arriba cubren el camino feliz. Esto cubre los archivos que de verdad llegan a producción.

Detectar el encoding

UTF-8 es el default correcto, pero Excel en Windows exporta en cp1252 y las exportaciones viejas vienen en latin-1. Prueba UTF-8 primero y luego haz fallback:

def open_csv_with_encoding(file_path):
    """Abre un CSV probando UTF-8 primero y luego encodings comunes de Windows."""
    for encoding in ("utf-8", "cp1252", "latin-1"):
        try:
            f = open(file_path, newline="", encoding=encoding)
            f.read(4096)  # sondea los primeros bytes antes de comprometerse
            f.seek(0)
            return f
        except UnicodeDecodeError:
            f.close()
    raise UnicodeDecodeError("unknown", b"", 0, 1, "no se pudo decodificar el archivo")

latin-1 decodifica cualquier cosa (todo byte es un carácter válido), así que siempre tiene éxito — exactamente por eso va último. Si el archivo resulta ser UTF-16 (las exportaciones “Texto Unicode” de Excel), chardet/charset-normalizer lo detecta de forma fiable.

Delimitadores distintos

Las locales europeas exportan CSV con ; porque la , es el separador decimal. Détéctalo en lugar de harcodearlo:

import csv

def detect_delimiter(file_path: str) -> str:
    """Detecta el delimitador más probable en un archivo CSV."""
    with open(file_path, "r", newline="", encoding="utf-8") as f:
        sample = f.read(1024)
        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
        return dialect.delimiter

delim = detect_delimiter("data.csv")
with open("data.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f, delimiter=delim)
    for row in reader:
        print(row)

El sniffing sobre una muestra pequeña falla de vez en cuando — los archivos de una sola columna no tienen nada que oler. Permite al usuario sobrescribir el delimitador detectado en tu interfaz de importación.

Múltiples hojas de Excel

import pandas as pd

def import_multi_sheet_excel(file_path: str) -> dict[str, list[dict]]:
    """Importa todas las hojas de un archivo Excel."""
    xl = pd.ExcelFile(file_path)
    result = {}
    for sheet_name in xl.sheet_names:
        df = pd.read_excel(file_path, sheet_name=sheet_name)
        df = df.dropna(how="all")  # descarta filas completamente vacías
        result[sheet_name] = df.to_dict("records")
    return result

Fechas de Excel

Excel almacena las fechas como números de serie (días desde 1900-01-01, menos su quirk de off-by-one). Parsea explícitamente o recibirás floats como 45789.0 donde debería haber un cumpleaños:

import pandas as pd

def import_excel_with_dates(file_path: str) -> list[dict]:
    """Importa Excel con parseo correcto de fechas."""
    df = pd.read_excel(
        file_path,
        parse_dates=["birth_date", "hire_date", "last_login"],
        date_format="%Y-%m-%d",
    )

    # Manejar formatos de fecha mixtos dentro de una columna
    for col in ["birth_date", "hire_date"]:
        df[col] = pd.to_datetime(df[col], errors="coerce", format="mixed")

    # Descartar filas donde una fecha requerida no se pudo parsear
    df = df.dropna(subset=["birth_date", "hire_date"])

    return df.to_dict("records")

errors="coerce" convierte los valores no parseables en NaT en lugar de lanzar excepción — combínalo con dropna (para rechazar) o con tu recolector de errores (para reportar). format="mixed" requiere pandas 2.0+.

Reanudar una importación interrumpida

Una importación de 2 horas que se rompe en la fila 49.999 no debería empezar de cero. Rastrea el progreso con un archivo de checkpoint:

import csv
import json
from pathlib import Path

def resumable_import(file_path: str, checkpoint_path: str, batch_size: int = 1000):
    """Importa CSV con checkpointing para capacidad de reanudación."""
    checkpoint = {"last_row": 0, "inserted": 0}

    if Path(checkpoint_path).exists():
        with open(checkpoint_path) as f:
            checkpoint = json.load(f)
        print(f"Reanudando desde fila {checkpoint['last_row']}")

    with open(file_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        batch = []

        for row_num, row in enumerate(reader, start=2):
            if row_num <= checkpoint["last_row"]:
                continue  # Saltar filas ya procesadas

            batch.append(row)

            if len(batch) >= batch_size:
                # Insertar lote a base de datos...
                checkpoint["last_row"] = row_num
                checkpoint["inserted"] += len(batch)

                with open(checkpoint_path, "w") as f:
                    json.dump(checkpoint, f)

                batch.clear()

        # Procesar filas restantes
        if batch:
            checkpoint["last_row"] = row_num
            checkpoint["inserted"] += len(batch)
            with open(checkpoint_path, "w") as f:
                json.dump(checkpoint, f)

    return checkpoint

El checkpoint solo funciona porque las filas se procesan en orden de archivo. Para colas de trabajo donde un worker cae a mitad de lote, ancla el checkpoint al último lote confirmado, no a las filas vistas — la misma idea que un processed_offset en un pipeline de streaming.

Importar a una base de datos

El INSERT fila por fila es donde mueren la mayoría de las importaciones: 50.000 round-trips a ~2ms cada uno son dos minutos de nada más. Hazlo por lotes.

import sqlite3

BATCH_SIZE = 1000

def batch_insert_users(db_path: str, rows: list[dict]):
    """Inserta filas validadas por lotes dentro de una transacción."""
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("BEGIN")
        for i in range(0, len(rows), BATCH_SIZE):
            chunk = rows[i:i + BATCH_SIZE]
            conn.executemany(
                "INSERT INTO users (name, email, age) VALUES (:name, :email, :age)",
                chunk,
            )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

Para cualquier cosa no trivial, haz staging primero: carga las filas crudas en una tabla temporal users_staging, valida ahí, y luego mueve los datos limpios con un solo INSERT INTO users SELECT ... FROM users_staging. Obtienes una vista previa de lo que va a aterrizar, el rollback es trivial (DROP TABLE) y la tabla de producción nunca se bloquea durante la validación. Los duplicados se resuelven con INSERT ... ON CONFLICT (PostgreSQL), INSERT IGNORE (MySQL) o MERGE (SQL Server) — elige una business key como email y reporta al usuario cuántos se saltaron.

Lo que funciona

  • Valida antes de insertar. Nunca confíes en archivos subidos por usuarios; comprueba cada celda contra tipos y rangos esperados — la misma disciplina que la validación de entrada en el borde de una API.
  • Reporta todos los errores de una vez, no solo el primero. Quien corrige un CSV quiere la lista completa en un intento, no el ensayo-error fila por fila. Devuelve un informe de errores descargable con números de fila y los valores problemáticos.
  • Envuelve las importaciones a base de datos en una transacción. Cuando una fila falla la validación, hacer rollback del lote gana a dejar una tabla a medio importar.
  • Soporta más de un encoding. Prueba UTF-8 primero y luego haz fallback a latin-1 o cp1252 para archivos legacy de Windows (ver el snippet de detección arriba).
  • Ofrece un archivo de plantilla. Un .xlsx/.csv descargable con los headers exactos y el formato que esperas elimina el fallo de importación más común antes de que ocurra.

Errores comunes

  • Insertar fila por fila. Los INSERT individuales son unas 100x más lentos que los inserts por lotes; usa executemany (Python), bulkCreate (Sequelize) o JDBC batch inserts.
  • Ignorar problemas de encoding. Un archivo con nombres en español guardado en Windows se corrompe o falla si fuerzas UTF-8 — detéctalo o deja que el usuario elija.
  • No manejar filas en blanco. Los archivos Excel suelen arrastrar cientos de filas vacías al final, así que filtra las filas donde todas las celdas están vacías (dropna(how="all"), no dropna() a secas que mata las filas parcialmente rellenas).
  • Sin rate limiting en subidas. Un Excel de 500MB tumbará la mayoría de los procesos de importación; impón tamaños máximos y empuja las importaciones grandes a trabajos en segundo plano.
  • Pérdida de datos silenciosa. Truncar columnas VARCHAR(255) sin avisar pierde datos — valida las restricciones de longitud explícitamente y reporta el truncado.
  • df.dropna() sin how. El comportamiento por defecto descarta cualquier fila con cualquier NaN, así que una celda en blanco mata un registro entero. how="all" es casi siempre lo que quieres.

Lectura adicional

Véase también

Preguntas frecuentes

¿Cómo manejo un archivo CSV de 1GB?

Nunca lo cargues completamente en memoria. Usa parsers de streaming (csv.reader en Python, csv-parser en Node.js, Apache Commons CSV en Java). Procesa una fila a la vez, valídala e insértala en lotes de 1.000-10.000 filas. Si la importación tarda más de unos segundos, descárgala a un background job y reporta el progreso de forma asíncrona.

¿Debo importar directamente a la base de datos o hacer staging primero?

Haz staging primero para cualquier cosa no trivial. Importa a una tabla temporal "staging", valida todo y luego ejecuta un INSERT INTO ... SELECT para mover los datos limpios a producción. Esto permite previsualizar errores, hacer rollback fácilmente y evitar bloquear tablas de producción durante la validación.

¿Cómo manejo filas duplicadas?

Define una business key (ej. email o SKU) y usa INSERT ... ON CONFLICT (PostgreSQL), INSERT IGNORE (MySQL) o MERGE (SQL Server, Oracle). Alternativamente, deduplica en memoria con un Set de hashes antes de insertar. Informa siempre al usuario de cuántos duplicados se encontraron y omitieron.

¿Debería usar CSV o Excel para importaciones?

CSV para pipelines programáticos y cualquier cosa grande — hace streaming, diferea limpio y cualquier lenguaje lo parsea. Excel para usuarios de negocio que preparan datos a mano — es lo que tienen, y preserva tipos que el CSV aplana a strings. Soporta ambos si la audiencia es mixta; internamente normaliza al mismo esquema de fila para que el código posterior no se entere.

¿Qué encoding debería asumir para archivos CSV?

Asume UTF-8, pero nunca lo exijas en silencio. Prueba UTF-8 primero y haz fallback a cp1252/latin-1 (ver el snippet de detección arriba) — o expone un selector de encoding en el formulario de subida. Escribir el encoding detectado en el informe de errores le ahorra una sesión de depuración a la siguiente persona.

¿Puedo importar Excel sin pandas?

Sí. openpyxl lee .xlsx directamente en Python puro, y hace streaming con read_only=True — mejor que read_excel para archivos que pandas no puede meter en memoria. En JavaScript, xlsx (SheetJS) hace lo equivalente; en Java, Apache POI es el estándar.