CSV 流式解析校验管道

示例说明

导入用户上传的 CSV 时，整文件读入再校验会把内存和延迟都放大。流式管道让解析器按行吐出对象，transform 环节做逐行校验：非法行记入错误报告后丢弃，合法行规范化后流向下游入库环节。上传多大的文件，内存占用都恒定在单行级别。

技术要点

• columns 与 cast 把原始行变结构化对象
• 逐行校验错误收集不中断整条流
• 异步生成器作为终端消费环节

代码解读

parse 选项里 columns 以首行为键名、cast 自动数字转换、bom 去除 Excel 文件的字节序标记。validator 对邮箱格式与年龄区间做检查，问题行推进 errors 数组并 callback(null) 跳过；sink 生成器逐条消费合法记录模拟入库动作。

使用建议

错误报告应保留行号便于用户修正后重传。数值列要显式验证 NaN。超宽文件设置 relax_column_count 或直接拒绝，防止字段错位产生脏数据。

代码示例

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);