中文
devexamples

CSV Parse Streaming Transform Pipeline

Parse CSV as a stream with csv-parse, validate each row against a schema, and pipe valid records onward while collecting row-level errors.

JavascriptNodeFile Upload

Code

snippet
import { parse } from 'csv-parse';
import { transform } from 'stream-transform';
import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';

const errors = [];
let validCount = 0;

const parser = parse({
  columns: true,
  skip_empty_lines: true,
  trim: true,
  cast: true,
  bom: true,
});

const validator = transform((row, callback) => {
  const issues = [];
  if (!row.email || !String(row.email).includes('@')) issues.push('bad email');
  if (!(row.age >= 0 && row.age <= 130)) issues.push('age out of range');

  if (issues.length > 0) {
    errors.push({ row: row.email ?? 'unknown', issues });
    return callback(null);
  }
  validCount += 1;
  callback(null, { email: String(row.email).toLowerCase(), age: row.age });
});

await pipeline(
  createReadStream('subscribers.csv', 'utf8'),
  parser,
  validator,
  async function* sink(records) {
    for await (const record of records) {
      console.log('upserting', record.email);
    }
  },
);

console.log('valid:', validCount, 'rejected:', errors.length);

Related Examples