Summing CSV files row by row using Node.js events

Question

I need to read multiple CSV files that have the same number of rows, and compute some results for each row. For simplification purposes, and in the scope of this code review, this computation would be "sum each column of each row".

I would like to be able to do so using the fast-csv parser that triggers events on each row reading. Mostly to avoid having to load the entire file into variables and then compute what I need.

I came up with the following solution, with a dummy example to sum the values from each file. It seems to be working, but because I'm working with events, I'm not sure if there might be a case when some data could go missing, or if there's any other issue. I would appreciate any help to improve performance, functionality, maintainability, readability, etc..

Because executing the code requires an external dependency and some files, here's a Repl.it.

Here's the code if you just want to read:

const csv = require("fast-csv");
const fs = require('fs');
const EventEmitter = require('events'); 

const files = ['csv/file1.csv', 'csv/file2.csv'];

// Create parsers for each file
const parsers = files.map(file => csv.parseStream(fs.createReadStream(file), { delimiter: ';', headers: false }));

// Variables to hold temporary data and definitive results
const data = parsers.map(() => []);
let result = [];

// Handler for a set of the same row from all the different files
var eventEmitter = new EventEmitter(); 
eventEmitter.on('allRows', (rows) => { 
  var sum = (r, a) => r.map((b, i) => Number(a[i]) + Number(b));
  result.push(rows.reduce(sum));
}); 

// Handler to display the result once all files have been read
eventEmitter.on('end', (rows) => { 
  console.log("result", result);
}); 

// Handler for each row reading
const onDataHandler = function(row, idx) {
  data[idx].push(row);
  // When we have a row parsed for each file, we emit a signal 
  if (data.reduce((acc, el) => acc && el.length > 0, true)) {
    if (data.reduce((acc, el) => acc && el[0] === null, true)) {
      // 'end' if all results have been shifted
      eventEmitter.emit('end'); 
    } else {
      // 'allRows' otherwise, with the first element of each file
      eventEmitter.emit(
        'allRows', 
        data.reduce((acc, el) => {
          acc.push(el.shift());
          return acc;
      }, [])
      ); 
    }
  }
}

// We use readable to use 'flowing' mode and make sure we don't miss the last rows
parsers.forEach((parser, idx) => parser.on('readable', () => {
  onDataHandler(parser.read(), idx);
}));

Answer 1

If you want to check whether every item in an array passes a test, you should use .every for that, not .reduce. Eg

data.reduce((acc, el) => acc && el.length > 0, true)

can be switched with

data.every(arr => arr.length)

and

data.reduce((acc, el) => acc && el[0] === null, true)

can be switched with

data.every(arr => arr[0] === null)

Also, when you want to transform one array into another, .map is the right method, not .reduce. Or, when you just want to copy an array exactly, use .slice() or spread it. Change

data.reduce((acc, el) => {
  acc.push(el.shift());
  return acc;
}, [])

to

[...data]

---

On a broader note, I would consider the logic to be a lot easier to follow if it was separated into two parts: one part to read the files and deal with the asynchronicity, and another part to (synchronously) combine into a single array. I think mixing the two and using eventEmitter makes things more difficult to understand than they need to be.

A somewhat minor issue is that you're requiring all rows of a particular index to be kept in memory until they've all been parsed. In the case that you're dealing with a large number of files, or the rows are large, it would be better to combine them as soon as possible, as they come in, so that the only persistent data structure is the output array of arrays (rather than an array of arrays for each separate .csv).

When any row is parsed, you can call a function that combines it with the results array. You can also map each parseStream to a Promise that resolves once its end event fires, and use Promise.all to log the final results array once all parsers are completed:

const result = [];
const combineRow = (row, rowIndex) => {
  const targetRow = result[rowIndex];
  if (!targetRow) {
    result[rowIndex] = row;
    return;
  }
  row.forEach((cell, cellIndex) => {
    targetRow[cellIndex] = (targetRow[cellIndex] || 0) + cell;
  });
}
const getRows = parser => new Promise((resolve, reject) => {
  let rowIndex = 0;
  parser.on('readable', () => {
    const row = parser.read();
    if (row) {
      combineRow(row.map(Number), rowIndex);
      rowIndex++;
    }
  });
  parser.on('end', resolve);
});

Promise.all(parsers.map(getRows))
  .then(() => {
    console.log(result);
  });

https://repl.it/repls/EnchantedElaborateFiber