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:
- Parsear el archivo fila por fila. El streaming importa —
csv.DictReader,csv-parseryCSVParseriteran sin cargar el archivo entero en memoria. - 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. - 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.
- 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
| Formato | Librería | ¿Streaming? | Ideal para |
|---|---|---|---|
| CSV | Python csv | Sí | Archivos pequeños a medianos con validación propia |
| CSV | csv-parser (JS) | Sí | Pipeline de streaming en Node.js |
| CSV | Apache Commons CSV | Sí | Parseo Java enterprise |
| Excel | pandas.read_excel | No | Importaciones rápidas, exploración de datos |
| Excel | xlsx (JS) | No | Cliente o importaciones pequeñas en servidor |
| Excel | Apache POI | Sí (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-1ocp1252para archivos legacy de Windows (ver el snippet de detección arriba). - Ofrece un archivo de plantilla. Un
.xlsx/.csvdescargable 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
INSERTindividuales son unas 100x más lentos que los inserts por lotes; usaexecutemany(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"), nodropna()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()sinhow. 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
- pandas IO tools —
read_csv/read_excel— la matriz completa de opciones, incluidosusecols,dtypey lectura por chunks. - Apache Commons CSV — parseo CSV en streaming para la JVM.
- Documentación de SheetJS — opciones de
xlsxpara parsear y escribir hojas de cálculo en JavaScript. - RFC 4180 — formato CSV — la definición oficial del formato; explica las reglas de entrecomillado que tu parser debe manejar.
- Repositorio complementario — los scripts de importación ejecutables más un archivo de muestra para probar.
Véase también
- Parsear archivos CSV — el camino de solo lectura cuando no necesitas validación ni persistencia.
- Parsear archivos Excel — lectura ligera de Excel sin la maquinaria de importación.
- Validación de entrada — los mismos patrones de validación aplicados a nivel de fila, aplicados a payloads de peticiones.
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.
Recursos Relacionados
Exportar Datos a CSV y Excel
Cómo exportar datos estructurados a archivos CSV y Excel de forma eficiente.
RecipeValidación de Subida de Archivos
Cómo manejar subidas de archivos de forma segura con validación de tamaño, tipo y contenido.
RecipeValidación de Input: Schemas en 3 Lenguajes
Valida el input de usuarios en el límite de la API con librerías de schemas. Ejemplos con Pydantic, Zod y Jakarta Validation, sanitización y errores estructurados.
RecipeAnalizar Archivos CSV
Cómo analizar archivos CSV en Python, Java y JavaScript con ejemplos de código prácticos.
RecipeAnalizar Archivos Excel en Python, JavaScript y Java
Cómo leer y escribir archivos Excel (.xlsx) en Python, JavaScript y Java con openpyxl, SheetJS y Apache POI.
RecipeTareas en Segundo Plano (Background Jobs)
Descarga trabajo lento de las peticiones HTTP: ejecuta tareas en segundo plano con colas de trabajo, programación cron y workers en Python, JavaScript y Java.