Parsing a CSV using a stream pipeline (async)

csv-parse-async.mjs

import { parse } from 'csv-parse'
import fs from 'fs'
import { pipeline } from 'stream/promises'

/**
 * Do something async with a row.
 *
 * @param {*} row A row of the CSV as an object.
 */
async function handleRow(row) {
  // Do something async
  console.log('Row', row)
}

/**
 * Read the CSV using a stream and, for example, return the number of rows handled.
 *
 * @param {string} filePath The full path to the CSV file to stream.
 */
async function readCsv(filePath) {
  let count = 0

  await pipeline(
    fs.createReadStream(filePath),
    parse({
      skip_empty_lines: true,
      columns: true
    }),
    async function* (source) {
      for await (const chunk of source) {
        yield await handleRow(chunk)
        count++
      }
    }
  )

  return count
}

;(async function main() {
  const count = await readCsv('path/to/file.csv')
  
  console.log(`FINISHED ${count} rows`)
})().catch((err) => {
  console.error('===== FATAL ERROR =====')
  console.log(new Date())
  console.error('Error', err)
  process.exit(1)
})