FICHERO_VENTAS lleva declarado desde la lección anterior apuntando a datos/ventas.csv, un fichero que aún no existe. Ha llegado su momento, y con él el problema que ninguna de las herramientas vistas hasta ahora puede resolver.

El catálogo de Escena Viva ocupa dos kilobytes: readFile lo lee entero sin despeinarse. El histórico de ventas de una temporada es otra cosa. Tres salas, doscientas sesiones al año, miles de entradas por sesión, cada una con su código, su canal de compra y su marca de tiempo: cientos de megabytes. Y readFile no tiene forma de leer eso sin meterlo entero en memoria.

Los streams son la respuesta de Node a ese problema, y son mucho más que una optimización: son el modelo con el que Node piensa la entrada y la salida. Las peticiones y respuestas HTTP del Módulo 4 son streams; la compresión, la criptografía y las conexiones de red son streams. Y —esto va a sonarte— los streams son EventEmitter, así que todo lo de la lección 02-05 se aplica aquí directamente.

1. El problema: medir lo que cuesta readFile
2. Qué es un stream y sus cuatro tipos
3. Los streams son EventEmitter
4. Modo fluido y modo pausado
5. createReadStream, createWriteStream y highWaterMark
6. Contrapresión: lo que devuelve write()
7. pipe() y su punto débil
8. El fichero datos/ventas.csv de Escena Viva
9. Leer línea a línea con readline
10. for await...of sobre un stream

No hay que creerse esto de palabra. Vamos a medirlo.

Con un ventas.csv de 800 MB, la salida es demoledora:

Y si el fichero fuera algo mayor, readFile ni siquiera llegaría a imprimir: fallaría con ERR_STRING_TOO_LONG —una cadena de V8 tiene un límite de unos 512 MB— o con JavaScript heap out of memory, que mata el proceso entero. La diferencia es de dos órdenes de magnitud, y la explicación es simple: readFile necesita tener el fichero completo en memoria a la vez, mientras que el stream lo procesa en trozos de 64 KB y descarta cada uno en cuanto lo ha usado. El consumo del stream no depende del tamaño del fichero: esa es la propiedad clave, memoria constante.

Un stream es una secuencia de datos que se procesa por partes, a medida que están disponibles, en lugar de esperar a tenerlos todos. La analogía del agua es buena: readFile es llenar la bañera entera antes de tocar el agua; un stream es abrir el grifo y trabajar con lo que va saliendo.

Los Transform son el tema de la lección siguiente. Aquí nos concentramos en los dos primeros, que son la base de todo lo demás.

Esto no es una analogía: Readable y Writable heredan de EventEmitter. on, once, emit, off y todo lo de la lección 02-05 funciona tal cual, incluida la regla de oro que ahora cobra su verdadero sentido.

Dos advertencias que valen por media lección. La primera: end y finish no son lo mismo. end es del lado que lee ("se acabó la entrada") y finish es del lado que escribe ("he terminado de volcar"); confundirlos lleva a cerrar un fichero antes de que sus datos estén en disco. La segunda: el evento error de un stream sigue la regla del EventEmitter que ya conoces —un error sin oyentes lanza una excepción no capturada y mata el proceso—. Un stream sin on('error') es una caída de producción esperando a ocurrir, y el caso más habitual no es un disco roto: es un ENOENT porque el fichero no estaba.

Un Readable tiene dos formas de entregar datos, y esta dualidad es la fuente de la mitad de los problemas de quien empieza con streams.

No los mezcles. Suscribirse a data y llamar a read() en el mismo stream produce comportamientos difíciles de razonar: trozos que se pierden, end que no llega, orden alterado. Elige un modo por stream y mantenlo —aunque en la práctica la recomendación es no usar ninguno directamente: pipe, pipeline o for await...of resuelven el problema mejor—. Y un tercer detalle del modo fluido: si te suscribes a data después de un await, puedes perder trozos, porque el stream empezó a fluir antes de que llegara tu oyente. Suscribe siempre de forma síncrona, justo después de crear el stream.

El highWaterMark es el tamaño del búfer interno: cuántos bytes acumula el stream antes de considerar que ya tiene bastante. Bajarlo (16 KB) gasta menos memoria por stream a cambio de más llamadas al sistema y más eventos; subirlo (1 MB) hace lo contrario. Los 64 KB por defecto son razonables casi siempre: bajarlo tiene sentido cuando manejas miles de streams simultáneos —un servidor con muchas conexiones— y subirlo cuando procesas pocos ficheros muy grandes. En modo objeto (que verás en la lección siguiente) el highWaterMark cuenta objetos, no bytes, y vale 16 por defecto.

Una advertencia sobre encoding: si no lo indicas, los trozos llegan como Buffer; y si lo indicas como 'utf8', el stream se encarga de que ningún carácter multibyte quede partido entre dos trozos —un problema real que veremos a fondo en la lección 03-06 con StringDecoder—.

Aquí está el concepto que separa a quien usa streams de quien los entiende. Imagina que lees de un SSD rápido y escribes en un disco lento o en una conexión de red: los datos entran a 500 MB/s y salen a 50 MB/s. ¿Dónde va la diferencia? Al búfer interno del stream de escritura, que crece sin parar hasta agotar la memoria. Es decir: has cambiado un readFile que consumía 800 MB por un stream que consume... 800 MB.

La contrapresión (backpressure) es el mecanismo que impide eso, y se apoya en un detalle que casi todo el mundo ignora:

write() devuelve un booleano. Y ese booleano no es informativo: es una orden.

Ese código funciona con ficheros pequeños y revienta con los grandes: el búfer del escritor crece indefinidamente porque nadie deja de alimentarlo. Es el fallo más común al escribir streams a mano, y el más difícil de diagnosticar, porque en desarrollo —fichero pequeño, disco rápido— no se manifiesta nunca.

El ciclo completo es: data → write() devuelve false → pause() → el búfer se vacía y el escritor emite drain → resume() → vuelta a data. El resultado es que el consumidor lento marca el ritmo del productor rápido, y la memoria se mantiene acotada por el highWaterMark. Esta idea no es exclusiva de los ficheros: es la misma que gobierna TCP y la que hará que un cliente lento no tumbe tu servidor HTTP en el Módulo 4.

Escribir a mano el baile de pause/resume/drain en cada tubería sería insoportable. pipe() lo hace por ti:

Esa línea equivale a todo el bloque anterior: conecta ambos streams, gestiona la contrapresión y llama a escritor.end() cuando el lector termina. Se pueden encadenar —a.pipe(b).pipe(c)— porque pipe devuelve el stream de destino.

Pero pipe tiene un defecto grave y poco conocido: no propaga errores ni limpia lo que deja atrás. Si lector falla —un ENOENT, un disco con errores—, el error no llega a escritor, que queda abierto con su descriptor sin liberar; y si nadie escucha error en lector, el proceso muere.

Con una tubería de tres o cuatro etapas el problema se multiplica: hay que suscribir on('error') en cada stream y cerrar a mano los que queden colgando. Es tedioso, es fácil olvidarse de uno, y ese olvido se paga en descriptores filtrados hasta el EMFILE.

Por eso la recomendación moderna es rotunda: usa pipe() para entender el concepto y stream.pipeline para trabajar. pipeline propaga errores, destruye todos los streams de la cadena cuando algo falla y tiene versión con promesas. Es el tema central de la lección siguiente.

Necesitamos datos con los que trabajar. Este es el formato del histórico de ventas:

Las columnas siguen las convenciones del proyecto: codigoEntrada con el formato EV-<año>-<6 dígitos>, sesionId con la forma ses-NNN-M, fechaVenta como cadena ISO sin zona, precioCentimos entero y canal con uno de tres valores (web, taquilla, app). Generamos un fichero coherente con la semilla: exactamente 1811 ventas, repartidas según las vendidas de cada sesión.

El comentario sobre la contrapresión es deliberado: con 1811 líneas no hay problema, pero el código correcto para volúmenes reales tendría que respetar el retorno de write() o, mejor, construirse con pipeline y un generador, como haremos en la lección siguiente.

Un CSV se procesa por líneas, pero los trozos de un stream no coinciden con las líneas: un trozo de 64 KB corta por la mitad la línea que le toque. En el laboratorio del apartado 1 lo resolvimos a mano con una variable resto; el módulo node:readline lo hace bien y sin código propio.

Tres decisiones dignas de mención. crlfDelay: Infinity evita que un fichero generado en Windows produzca líneas terminadas en \r invisible que estropea el último campo. Una línea corrupta se registra y se salta, no aborta el informe: en un histórico de millones de registros siempre hay basura, y un informe que muere en la línea 4.000.000 no sirve de nada. Y el consumo de memoria es constante: porSesion guarda una entrada por sesión (siete), no por venta.

Ese for await (const linea of lector) merece explicación, porque es la forma más legible de consumir un stream y ya la conoces de la lección 02-05, cuando viste events.on().

Todo Readable es un iterable asíncrono. Se puede recorrer directamente:

Sus ventajas sobre on('data') son concretas: la contrapresión es automática (el bucle no pide el siguiente trozo hasta terminar el cuerpo) en lugar de manual; los errores se capturan con un try/catch normal alrededor del bucle en lugar de con un on('error') aparte; se puede usar await dentro; y un break destruye el stream solo, sin destroy() explícito.

Ese await dentro del bucle es la diferencia decisiva. Si por cada línea del CSV hay que consultar una base de datos, con on('data') las consultas se lanzarían todas a la vez y hundirían el servidor; con for await, la lectura se detiene sola hasta que la consulta termina, así que la contrapresión sale gratis. El precio es rendimiento bruto: for await es algo más lento porque cada iteración crea una promesa. Para un proceso por lotes es irrelevante; en una ruta crítica con mucho tráfico, mídelo.

• No suscribir on('error'). Un error sin oyentes en un EventEmitter mata el proceso, y un stream es un EventEmitter. La solución definitiva es pipeline.
• Ignorar el retorno de write(). Funciona en desarrollo y agota la memoria en producción. Si escribes a mano, respeta el drain.
• Confundir end con finish. end es el lector agotado; finish es el escritor vaciado. Cerrar por el evento equivocado corta datos.
• Suponer que un trozo es una línea. Nunca lo es. Usa readline o un Transform que trocee.
• Mezclar modo fluido y pausado, o suscribirse a data tras un await. En el primer caso el comportamiento es errático; en el segundo pierdes trozos.
• Acumular todos los trozos en un array para juntarlos al final. Eso es readFile con pasos extra: has perdido la única ventaja del stream.
• Consejo: si el fichero cabe holgadamente en memoria y solo lo lees una vez, readFile es más simple y más rápido; los streams son para lo grande, lo infinito o lo que llega poco a poco. Y cuando dudes de si hay contrapresión, imprime escritor.writableLength de vez en cuando: si crece sin parar, la estás ignorando.

Amplía src/informes/ventas-por-sesion.js con una función resumirPorCanal() que recorra datos/ventas.csv una sola vez y devuelva, por canal, el número de entradas, la recaudación total y la hora punta (la hora del día con más ventas). Debe usar readline, mantener memoria constante y validar que la suma de entradas de todos los canales es 1811.

Escribe src/laboratorio/copiar-con-contrapresion.js que copie un fichero grande sin usar pipe, respetando el drain, e imprima cada segundo: megabytes copiados, writableLength del escritor y número de pausas por contrapresión. Ejecuta después una versión que ignore el retorno de write() y compara el consumo de memoria con process.memoryUsage().rss.

Escribe src/laboratorio/filtrar-ventas.js que lea datos/ventas.csv, se quede solo con las ventas cuyas sesiones pertenezcan a una sala dada (--sala="Sala Boveda", cruzando con el catálogo) y escriba el resultado en datos/ventas-<sala>.csv conservando la cabecera. Usa readline para leer y un createWriteStream para escribir, respetando la contrapresión.

Solución 1. La clave es acumular en dos niveles durante el mismo recorrido:

El array de 24 posiciones por canal es el truco que mantiene la memoria constante: no se guardan las ventas, solo el contador de cada hora. Un Map de fechas completas crecería con el fichero.

Solución 2. El núcleo de la copia manual:

Con la contrapresión respetada, writableLength oscila alrededor del highWaterMark y el rss se mantiene plano. Sin ella, writableLength crece sin límite y el rss sube hasta acercarse al tamaño del fichero. Fíjate también en los dos on('error') cruzados que hacen falta para no filtrar descriptores: ese trabajo manual es exactamente lo que pipeline automatiza.

Solución 3. Lo importante es cruzar el catálogo antes de empezar a leer el CSV, para no hacerlo por línea:

Ese await new Promise(... 'drain' ...) es la traducción de la contrapresión al mundo de async/await, y es un patrón que merece la pena memorizar: dentro de un for await, esperar el drain detiene también la lectura, porque el bucle no pide el siguiente elemento hasta que termina el cuerpo. Un Set con los ids de sesión evita recorrer el catálogo 1811 veces.

Ya sabes por qué existen los streams y qué problema resuelven exactamente: has medido la diferencia entre 1621 MB de readFile y 6 MB de un stream sobre el mismo fichero, y entiendes que el consumo de un stream no depende del tamaño de la entrada. Conoces los cuatro tipos —Readable, Writable, Duplex y Transform— y el hecho fundamental de que son EventEmitter, con data, end, error, close, finish y drain, sin confundir end con finish y sabiendo que un error sin oyentes tumba el proceso.

Distingues el modo fluido del pausado y por qué no deben mezclarse; sabes qué es el highWaterMark y qué se gana y se pierde al moverlo. Y, sobre todo, entiendes la contrapresión: que write() devuelve un booleano que es una orden, que ignorarlo convierte tu stream en un readFile disfrazado, y que el ciclo pause → drain → resume es lo que hace que el consumidor lento marque el ritmo. Sabes que pipe() automatiza ese baile pero no propaga errores ni limpia descriptores, y por eso no es la herramienta final. Escena Viva, mientras tanto, tiene ya su datos/ventas.csv con las 1811 ventas de la semilla y un src/informes/ventas-por-sesion.js que lo recorre línea a línea con readline, acumulando entradas, recaudación y canales por sesión con memoria constante. Y has visto la forma más legible de consumir cualquier stream: for await...of, que trae la contrapresión de regalo y permite await dentro del bucle.

En Streams de Transformación y pipeline cerramos el círculo. Aprenderás a escribir tus propios streams con Transform (_transform y _flush), a montar tuberías de proceso completas —CSV crudo a objetos, filtrado por sala, agregación por sesión, escritura del informe—, y a sustituir pipe() por stream.pipeline, que propaga errores, destruye toda la cadena cuando algo falla y tiene versión con promesas. Verás además Readable.from(), los generadores asíncronos como transformaciones, la compresión con zlib intercalada en la tubería y la cancelación con AbortSignal.