Cómo ejecutar una función asíncrona para cada línea de un archivo muy grande (> 1 GB) en Node.js

Question

Digamos que tiene una gran (> 1 GB) CSV de ID de registros:

655453 
4930285 
493029 
4930301 
493031 
... 

Y para cada id desea realizar una llamada a la API REST para recuperar los datos del registro, transformarlos localmente e insertarlos en una base de datos local.

¿Cómo se hace eso con Node.js 'Readable Stream?

Mi pregunta es básicamente esta: ¿cómo se lee un archivo muy grande, línea por línea, se ejecuta una función asíncrona para cada línea, y [opcionalmente] se puede comenzar a leer el archivo desde una línea específica?

A partir de la siguiente pregunta Quora estoy empezando a aprender a usar fs.createReadStream:

http://www.quora.com/What-is-the-best-way-to-read-a-file-line-by-line-in-node-js

var fs = require('fs'); 
var lazy = require('lazy'); 

var stream = fs.createReadStream(path, { 
    flags: 'r', 
    encoding: 'utf-8' 
}); 

new lazy(stream).lines.forEach(function(line) { 
    var id = line.toString(); 
    // pause stream 
    stream.pause(); 
    // make async API call... 
    makeAPICall(id, function() { 
    // then resume to process next id 
    stream.resume(); 
    }); 
}); 

embargo, que pseudocódigo no funciona, debido a que las fuerzas lazy module a leer todo el archivo (como una secuencia, pero no hay pausa). Entonces, ese enfoque no parece funcionar.

Otra cosa es, me gustaría poder comenzar a procesar este archivo desde una línea específica. La razón de esto es que procesar cada id (realizar la llamada de la API, limpiar los datos, etc.) puede llevar hasta medio segundo por registro, así que no quiero tener que comenzar desde el principio del archivo cada vez . El enfoque ingenuo que estoy pensando utilizar es capturar el número de línea de la última identificación procesada y guardar eso. Luego, cuando vuelva a analizar el archivo, transmita todos los ID, línea por línea, hasta que encuentre el número de línea que dejó en, y luego haga el negocio makeAPICall. Otro enfoque ingenuo es escribir archivos pequeños (digamos de 100 ids) y procesar cada archivo de a uno por vez (un conjunto de datos suficientemente pequeño para hacer todo en la memoria sin una secuencia IO). ¿Hay una mejor manera de hacer esto?

puedo ver cómo esto se complica (y donde node-lazy viene en) porque el chunk en stream.on('data', function(chunk) {}); sólo podrán contener parte de una línea (si el bufferSize es pequeño, cada bloque puede ser de 10 líneas, pero debido a que el id es longitud variable, solo puede ser 9.5 líneas o lo que sea). Es por eso que me pregunto cuál es el mejor enfoque para la pregunta anterior.

Answer 1

Supongo que no necesita usar node-lazy. Esto es lo que he encontrado en Node docs:

Evento: data

function (data) { } 

El evento data emite ya sea un Buffer (por defecto) o una string si setEncoding() se utilizó.

Lo que significa esto es que llame setEncoding() en su corriente, entonces su data caso de devolución de llamada aceptará un parámetro de cadena. Luego, dentro de esta devolución de llamada puede llamar a los métodos .pause() y .resume().

El pseudo código debería tener este aspecto:

stream.setEncoding('utf8'); 
stream.addListener('data', function (line) { 
    // pause stream 
    stream.pause(); 
    // make async API call... 
    makeAPICall(line, function() { 
     // then resume to process next line 
     stream.resume(); 
    }); 
}) 

Aunque los documentos no especifican explícitamente que la corriente se lee línea a línea supongo que ese es el caso para los flujos de archivos. Al menos en otros idiomas y plataformas, las secuencias de texto funcionan de esa manera y no veo ninguna razón para que las transmisiones de Node difieran.

Answer 2

relacionada con la respuesta de Andrew Андрей Листочкин:

Se puede utilizar un módulo de byline como para conseguir un evento separado data para cada línea. Es una secuencia de transformación alrededor de la extensión de archivos original, que produce un evento data para cada fragmento. Esto te permite pausar después de cada línea.

byline no leerá todo el archivo en la memoria como lazy aparentemente lo hace.

var fs = require('fs'); 
var byline = require('byline'); 

var stream = fs.createReadStream('bigFile.txt'); 
stream.setEncoding('utf8'); 

// Comment out this line to see what the transform stream changes. 
stream = byline.createStream(stream); 

// Write each line to the console with a delay. 
stream.on('data', function(line) { 
    // Pause until we're done processing this line. 
    stream.pause(); 

    setTimeout(() => { 
     console.log(line); 

     // Resume processing. 
     stream.resume(); 
    }, 200); 
});