Iteración eficiente sobre archivos grandes en Python

Cuando abres un archivo con open() y escribes for linea in archivo:, Python no carga el archivo completo en memoria. El objeto file implementa el protocolo iterador de Python: cada llamada a __next__() lee exactamente una línea del buffer del sistema operativo y la devuelve. El archivo puede pesar 50 GB; tu proceso verá picos de kilobytes.

Este comportamiento no es magia —es una consecuencia directa del modelo de datos de Python. Un objeto file implementa __iter__ (devuelve self) y __next__ (lee la siguiente línea hasta StopIteration en EOF). Eso lo convierte en un iterador perezoso (lazy iterator): produce valores bajo demanda, sin materializar la secuencia completa.

El problema aparece cuando usas archivo.readlines(). Ese método sí carga todo el contenido, construye una lista de strings en heap y la devuelve. Con un log de 8 GB en un servidor con 4 GB de RAM, el resultado es un MemoryError o, peor, un swap masivo que degrada todo el host. read() sin argumentos tiene el mismo problema. La diferencia entre for linea in f: y f.readlines() es exactamente la diferencia entre procesar y acumular.

Para control manual existe readline(): lee una línea y avanza el cursor interno del archivo. Útil cuando necesitas leer selectivamente (saltar cabeceras, retroceder con seek, parsear protocolos con líneas de longitud variable). Pero para archivos binarios —imágenes, dumps, streams de red— no hay concepto de “línea”. Ahí el patrón correcto es leer chunks de tamaño fijo con iter(lambda: f.read(8192), b''): crea un iterador que llama al lambda repetidamente hasta que el valor centinela b'' (EOF) aparece.

Si omites el centinela o usas el tamaño de chunk equivocado, terminas con bucles infinitos o con líneas que cruzan fronteras de chunk y corrompen tu parsing.

import csv
import hashlib
from pathlib import Path


# ── 1. Iteración línea a línea: logs grandes ─────────────────────────
def count_errors(log_path: Path) -> int:
    """Cuenta líneas que contienen ERROR sin cargar el archivo completo."""
    count = 0
    with open(log_path, encoding="utf-8") as f:
        for line in f:                      # f.__next__() una línea a la vez
            if "ERROR" in line:
                count += 1
    return count


# ── 2. readline() para control manual ────────────────────────────────
def skip_header_and_process(log_path: Path) -> list[str]:
    """Lee la primera línea como cabecera y procesa el resto."""
    results = []
    with open(log_path, encoding="utf-8") as f:
        header = f.readline()               # avanza el cursor una línea
        if not header.startswith("#"):
            raise ValueError("Formato inesperado: cabecera no encontrada")
        for line in f:                      # el cursor ya apunta a la línea 2
            results.append(line.strip())
    return results


# ── 3. Chunks binarios: hash de un archivo arbitrariamente grande ─────
def sha256_of_file(file_path: Path) -> str:
    """
    Calcula SHA-256 leyendo en bloques de 64 KB.
    iter(callable, sentinel) llama al callable hasta que devuelve sentinel.
    b'' es lo que f.read() devuelve al llegar a EOF.
    """
    digest = hashlib.sha256()
    chunk_size = 65_536                     # 64 KB: equilibrio entre syscalls y memoria

    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            digest.update(chunk)            # procesa el chunk sin acumularlo

    return digest.hexdigest()


# ── 4. CSV grande con csv.reader (también perezoso) ───────────────────
def sum_column(csv_path: Path, column_index: int) -> float:
    """
    csv.reader envuelve el iterador del archivo: sigue siendo perezoso.
    Nunca llames list(csv.reader(f)) en archivos de producción.
    """
    total = 0.0
    with open(csv_path, newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        next(reader)                        # descarta la fila de cabecera
        for row in reader:
            try:
                total += float(row[column_index])
            except (ValueError, IndexError):
                pass                        # fila malformada: continúa sin explotar
    return total


if __name__ == "__main__":
    base = Path(".")

    # Genera un log de prueba para demostrar count_errors
    sample_log = base / "sample.log"
    sample_log.write_text(
        "INFO start\nERROR disk full\nINFO ok\nERROR timeout\n",
        encoding="utf-8",
    )
    print(f"Errores encontrados: {count_errors(sample_log)}")  # → 2

    # Hash del propio script (archivo binario arbitrario)
    import sys
    script = Path(sys.argv[0])
    print(f"SHA-256: {sha256_of_file(script)}")

Qué está pasando en cada decisión

El bucle for line in f es la forma canónica porque delega el buffering al runtime de CPython (que usa io.BufferedReader internamente con un buffer de 8 KB por defecto). No tienes que pensar en tamaños; Python gestiona las llamadas a read() del sistema operativo de forma transparente. La línea entera llega a tu variable, incluido el \n final —por eso el .strip() aparece cuando necesitas texto limpio.

readline() antes del bucle en skip_header_and_process muestra algo importante: readline() y el iterador comparten el mismo cursor interno (f.tell()). Después de readline(), el for line in f: continúa exactamente donde lo dejó la lectura manual. No hay desincronización, porque ambos pasan por el mismo BufferedReader.

iter(lambda: f.read(chunk_size), b"") merece atención especial. La firma de iter() con dos argumentos es iter(callable, sentinel) —una forma que mucha gente no conoce. El lambda captura f y chunk_size por closure. El centinela b"" (bytes vacíos, no None, no 0) es exactamente lo que f.read() devuelve cuando llega al EOF en modo binario. El tamaño de 64 KB no es arbitrario: con chunks demasiado pequeños pagas overhead de syscalls; con chunks demasiado grandes presionas el GC con objetos grandes. Para hashing o procesamiento de streams, 32–128 KB es el rango razonable en hardware moderno.

csv.reader sigue siendo perezoso porque acepta cualquier iterable. Al pasarle f (que es un iterador), csv.reader llama a next(f) internamente por cada fila que parsea. La composición de iteradores es el patrón real aquí: fcsv.reader(f) → tu bucle. Ninguna capa materializa todo. Si en algún momento haces rows = list(csv.reader(f)) en un CSV de producción, has vuelto al problema de readlines().

Errores que debes conocer

Error: Mezclar readline() y for line in f: con io.TextIOWrapper en modo texto no causa problemas, pero hacerlo sobre un socket o un RawIOBase puede desincronizar el buffer interno y saltar líneas.

# ❌ Wrong: sobre un socket wrapeado manualmente
raw = socket.makefile("rb")
first = raw.readline()
for chunk in iter(lambda: raw.read(4096), b""):  # puede perder datos
    process(chunk)

# ✅ Right: elige UNA estrategia de lectura por objeto
raw = socket.makefile("rb")
for chunk in iter(lambda: raw.read(4096), b""):
    process(chunk)

El problema es que readline() puede haber consumido más bytes del socket en su buffer interno de los que devolvió. Mezclar las dos estrategias en el mismo objeto puede descartar esos bytes.


Error: Usar f.read() sin argumento dentro de un bucle pensando que lee incrementalmente.

# ❌ Wrong: carga TODO el archivo en cada iteración del bucle externo
with open("huge.bin", "rb") as f:
    while data := f.read():     # primer read() devuelve los 8 GB completos
        process(data)

# ✅ Right: pasa un tamaño explícito
with open("huge.bin", "rb") as f:
    while chunk := f.read(65_536):
        process(chunk)

f.read() sin argumento lee hasta EOF en una sola llamada. El walrus operator no cambia eso; solo evalúa si el resultado es truthy. Con el tamaño explícito, cada iteración procesa exactamente chunk_size bytes o menos (el último chunk).


El patrón iter(callable, sentinel) es uno de esos rincones de Python que, una vez que lo tienes interiorizado, reemplaza una cantidad sorprendente de bucles while True con break.

83

Dejar un comentario

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Scroll al inicio