El readStream que está utilizando aquí es asíncrono, lo que significa que .on(event, callback) se activará cada vez que se lea un nuevo dato, independientemente de cualquier callback de llamada activada. En otras palabras, la ejecución de la función de callback de llamada aquí no afecta este proceso, se ejecutará en paralelo cada vez que se reciba un event .

Esto significa que, en caso de que la callback de llamada fuera a ejecutar un fragmento de código que es asíncrono, es muy posible que termine en una situación en la que aún se estén ejecutando varias instancias de esta función en el momento en que se reciba el siguiente event lectura.

Nota: esto es válido para cualquier evento, incluido el evento 'end' .

Si tuviera que usar async/await en la callback de llamada, solo haría que la lógica interna de esta función fuera sincrónica. Todavía no afectaría la velocidad a la que se leen sus datos.

Para hacerlo, querrá usar async/await en la callback de llamada (para que sea sincrónico internamente) y hacer que la callback de llamada pause y reanude manualmente la operación de lectura en paralelo.

 const csv = require('csv-parser'); const fs = require('fs'); let i = 1; const stream = fs.createReadStream('table.csv').pipe(csv()); stream.on('data', async (row) => { // pause overall stream until this row is processed stream.pause(); // process row switch (i%4){ case 1: await org1createPatient(row.patientId, row.FirstName, row.LastName, row.Age, row.Sex, row.ChestPainType, row.RestingBP, row.Cholesterol, row.FastingBS, row.RestingECG, row.MaxHR, row.ExerciseAngina, row.Oldpeak, row.ST_Slope, row.HeartDisease); break; case 2: await org2createPatient(row.patientId, row.FirstName, row.LastName, row.Age, row.Sex, row.ChestPainType, row.RestingBP, row.Cholesterol, row.FastingBS, row.RestingECG, row.MaxHR, row.ExerciseAngina, row.Oldpeak, row.ST_Slope, row.HeartDisease); break; case 3: await org3createPatient(row.patientId, row.FirstName, row.LastName, row.Age, row.Sex, row.ChestPainType, row.RestingBP, row.Cholesterol, row.FastingBS, row.RestingECG, row.MaxHR, row.ExerciseAngina, row.Oldpeak, row.ST_Slope, row.HeartDisease); break; case 0: await org4createPatient(row.patientId, row.FirstName, row.LastName, row.Age, row.Sex, row.ChestPainType, row.RestingBP, row.Cholesterol, row.FastingBS, row.RestingECG, row.MaxHR, row.ExerciseAngina, row.Oldpeak, row.ST_Slope, row.HeartDisease); break; } i++; // resume overall stream stream.resume(); }); stream.on('end', () => { // now guaranteed that no instances of `callback` is still running in parallel when this event is fired console.log('CSV file successfully processed'); });

La solución a continuación es usar la biblioteca iter-ops , que es muy eficiente en este caso, porque pipe(csv()) devuelve un AsyncIterable , por lo que debe procesarse en consecuencia.

Como no le importa lo que devuelvan esas funciones de procesamiento, podemos limitar el procesamiento de cada fila:

 const {pipe, throttle, onEnd, catchError} = require('iter-ops'); const csv = require('csv-parser'); const fs = require('fs'); const asyncIterable = fs.createReadStream('table.csv').pipe(csv()); const i = pipe( asyncIterable, throttle(async (row, index) => { switch (index % 4) { case 1: await org1createPatient(row.patientId, ...); break; case 2: await org2createPatient(row.patientId, ...); break; case 3: await org3createPatient(row.patientId, ...); break; case 0: await org4createPatient(row.patientId, ...); break; default: break; } }), onEnd(s => { console.log(`Completed ${s.count} rows, in ${s.duration}ms`); }), catchError((err, ctx) => { console.log(`Failed on row with index ${ctx.index}:`, err); throw err; // to stop the iteration }) ); async function processCSV() { // this will trigger the iteration: for await(const a of i) { // iterate and process the CSV } }

PD: soy el autor de iter-ops .