csv-zod-stream

Streaming CSV/TSV parsing with per-row Zod validation — a real Node.js Transform on top of csv-parse, plus a Web Streams build for Deno, Bun and edge runtimes. Zod-first, and since 2.0 any Standard Schema will do: zod/mini, Valibot, ArkType.

import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
import { z } from 'zod';
import { createCsvValidator } from 'csv-zod-stream';

const Employee = z.object({
    id: z.coerce.number().int().positive(),
    name: z.string().min(1),
    email: z.email(),
    salary: z.coerce.number().int(),
});

await pipeline(
    createReadStream('employees.csv'),
    createCsvValidator(Employee),
    async function (rows) {
        for await (const employee of rows) await save(employee);
    }
);

• Backpressure that reaches the file handle. A slow consumer stops the reader, not the heap.
• A real Transform, so .pipe(), pipeline() and for await all work as you expect.
• Physical line numbers on errors, correct even when records span several lines.
• CSV, TSV, semicolon and pipe files, with an optional delimiter sniffer.
• Headers that fit the schema, folded to camelCase or snake_case, or renamed one by one.
• Files that are not UTF-8, decoded on the way in — windows-1251, latin1, whatever the export produced.
• Async validation, so a refinement can hit the database while backpressure still holds.
• Blank cells that behave, so .optional() and .nullable() work without a preprocess in every schema.
• csv-parse under the hood — quoting, escaping, BOM and CRLF are its problem, not a hand-rolled parser's.
• Zod v4 or v3, typed end to end: the row type is inferred from the schema, after coercion.
• Or any other Standard Schema — zod/mini for an edge bundle, Valibot, ArkType — with the same types and the same errors.
• Progress you can show, from stats on the stream or an onProgress callback.

Install

npm install csv-zod-stream zod

Node ≥ 20. zod is an optional peer dependency — ^3.24.0 || ^4.0.0, so a project already on zod 3.24+ keeps it and a fresh install gets v4. It is optional because any Standard Schema works; install whichever validator you use, or none of them if you pass a schema of your own. csv-parse comes along as a dependency.

Any Standard Schema

Everything below is written with Zod, and Zod needs no adapter. The package only ever asks a schema to validate one row, which is exactly the Standard Schema contract, so zod/mini (much smaller in a web or edge bundle), Valibot and ArkType work with no change at the call site:

import * as v from 'valibot';

const Employee = v.object({
    id: v.pipe(v.string(), v.transform(Number)),
    name: v.pipe(v.string(), v.minLength(1)),
});

createCsvValidator(Employee); // rows are typed v.InferOutput<typeof Employee>

A schema that validates asynchronously is awaited whether or not async is set — that option is about taking Zod's safeParseAsync route. zodError on a row error is only there for Zod; read issues instead, which every schema fills in.

Backpressure

Rows leave the parser only while the readable side has room for them. Stop reading and the pull loop stops, csv-parse fills up, its writes start returning false, and the upstream chunk callback is withheld — so a slow consumer throttles the file read instead of filling the heap.

npm run bench (500 000 rows, 8.3 MB, a consumer that yields every 1 000 rows):

    50000     7.9 MB  ██████████████████████████████████
   100000     9.3 MB  ████████████████████████████████████████
   150000     7.8 MB  ██████████████████████████████████
   200000     8.0 MB  ███████████████████████████████████
   250000     8.2 MB  ███████████████████████████████████
   300000     8.4 MB  ████████████████████████████████████
   350000     8.6 MB  █████████████████████████████████████
   400000     8.7 MB  ██████████████████████████████████████
   450000     8.9 MB  ██████████████████████████████████████
   500000     7.8 MB  █████████████████████████████████

Flat, not linear. The test suite asserts the same thing without measuring the heap: over a million rows, the producer never gets more than a couple of chunks ahead of the consumer.

Invalid rows

onInvalidRow decides what a row the schema rejects does to the stream.

'error' (default) — stop at the first one

try {
    await pipeline(createReadStream('employees.csv'), createCsvValidator(Employee), sink);
} catch (error) {
    if (error instanceof RowValidationError) {
        console.error(`line ${error.line}: ${error.message}`);
        console.error(error.raw);
    }
}

The stream is destroyed immediately, so rows still sitting in its buffer are dropped. Use 'skip' or 'collect' when you need everything up to the failure.

'skip' — drop it and carry on

const rows = createCsvValidator(Employee, { onInvalidRow: 'skip' });

rows.on('invalid-row', error => log.warn(`skipped line ${error.line}`, error.issues));

'collect' — drop it, and keep it for later

const rows = createCsvValidator(Employee, { onInvalidRow: 'collect', maxErrors: 100 });

await pipeline(createReadStream('employees.csv'), rows, sink);

for (const error of rows.errors) console.log(error.line, error.issues);

maxErrors is how many invalid rows to tolerate; the next one destroys the stream with a TooManyInvalidRowsError. It carries count, maxErrors, errors (the rows kept so far, under 'skip' as well as 'collect') and cause, the row that went over the line. It applies to 'skip' as well.

keepErrors — the ceiling on what is held in memory

A file that is bad end to end would otherwise put every rejected row and its ZodError on the heap, which is exactly what the streaming is there to avoid. keepErrors is the cap, 1000 by default; past it the oldest kept error is dropped and droppedErrors counts what fell off.

const rows = createCsvValidator(Employee, { onInvalidRow: 'collect', keepErrors: 50 });

await pipeline(createReadStream('employees.csv'), rows, sink);

console.log(`${rows.errors.length} kept, ${rows.droppedErrors} dropped`);

Pass Infinity for the old unbounded behaviour, or 0 to keep none. Under 'skip' the same window is what fills TooManyInvalidRowsError.errors; .errors and .droppedErrors stay empty and 0, since 'skip' collects nothing by design.

Anything thrown while a row is being judged — an onRowError that throws, a .transform() that throws, a schema with an async refinement that needs safeParseAsync — fails the stream with that error instead of escaping as an uncaught exception.

Row errors

errors, onRowError and the 'invalid-row' event all carry a CsvRowError: either a RowValidationError, from a row the schema rejected, or a RowParseError, from a record csv-parse could not read (see structural errors).

abstract class CsvRowError extends Error {
    line: number; // physical line the record starts on
    raw: string; // the record as it appeared in the file
}

class RowValidationError extends CsvRowError {
    record: number; // index among the data rows
    issues: readonly { path: readonly PropertyKey[]; message: string }[];
    zodError: ZodError | undefined; // only when the schema was Zod
}

class RowParseError extends CsvRowError {
    code: string; // the csv-parse error code
    cause: Error;
}

Narrow with instanceof before reaching for issues:

for (const error of rows.errors) {
    if (error instanceof RowValidationError) console.log(error.line, error.issues);
    else console.log(error.line, error.code);
}

issues is the same list whichever validator produced it: the path to the field and the message. zodError is the untouched ZodError when the schema was Zod, and undefined otherwise — reach for it when you want format() or a code, and check it first.

line is a file position, not a record count. Records holding multiline quoted fields push the two apart, which is exactly when a line number is worth having. Blank lines, comment lines and CRLF endings — including a CRLF inside a quoted field — are all accounted for. raw is the record's own text: the lines skipped ahead of it and its trailing line break are cut off.

A rejects file

An import is usually expected to hand back the rows it would not take. rejectsCsv renders the collected errors as a CSV:

import { writeFile } from 'node:fs/promises';
import { rejectsCsv } from 'csv-zod-stream';

const rows = createCsvValidator(Employee, { onInvalidRow: 'collect' });

await pipeline(createReadStream('employees.csv'), rows, sink);
await writeFile('rejects.csv', rejectsCsv(rows.errors));

line,record,error,code,field,message,raw
3,2,RowValidationError,,email,Invalid CSV row 2 at line 3 — email: Invalid email address,"2,ada,not-an-email,90"

record and field are filled in for a row the schema rejected, code for one csv-parse could not read. delimiter, eol and header are all overridable, and bom: true prepends a BOM so Excel opens the file as UTF-8.

Async validation

A schema with an async refinement or transform needs safeParseAsync, which async: true turns on:

const Employee = z.object({
    id: z.coerce.number(),
    email: z.string().refine(async email => !(await taken(email)), 'already registered'),
});

createCsvValidator(Employee, { async: true });

Rows are then validated one at a time, in file order, and the reader stops while a row is in flight — the same backpressure as the synchronous path, so an async lookup throttles the file instead of queueing a million promises. A rejected promise fails the stream with that error; an issue the refinement reports is an ordinary invalid row and obeys onInvalidRow.

Without async: true an async schema fails the stream with zod's own complaint about a synchronous parse, which is what it did before this option existed.

Blank cells

A blank cell is an empty string, which is not what .optional() or .nullable() are waiting for. emptyAs translates:

const Employee = z.object({
    id: z.coerce.number(),
    nickname: z.string().optional(),
    manager: z.string().nullable(),
});

createCsvValidator(Employee, { emptyAs: 'undefined' }); // '' -> undefined
createCsvValidator(Employee, { emptyAs: 'null' }); // '' -> null

'undefined' also lets .default() fire. Only a genuinely empty cell is translated — ' ' stays a space, unless trim is on.

Column names

Real files arrive with "First Name ", EMAIL and user_id where the schema says firstName. normalizeHeaders folds the header row before anything else looks at it:

createCsvValidator(Employee, { normalizeHeaders: 'camel' }); // 'First Name ' -> firstName
createCsvValidator(Employee, { normalizeHeaders: 'snake' }); // 'First Name ' -> first_name
createCsvValidator(Employee, { normalizeHeaders: 'lower' }); // 'EMAIL'       -> email
createCsvValidator(Employee, { normalizeHeaders: 'trim' }); //  ' email '     -> email
createCsvValidator(Employee, { normalizeHeaders: (header, index) => header || `column${index}` });

'camel' and 'snake' split on anything that is not a letter or a digit and on a camelCase boundary, so userID and User ID both land on userId or user_id. Letters outside ASCII are letters: Имя folds to имя, not to nothing.

columnAliases renames what folding cannot reach:

createCsvValidator(Employee, {
    normalizeHeaders: 'camel',
    columnAliases: { 'E-Mail': 'email', 'Annual salary': 'salary' },
});

An alias is looked up against the header as it appears in the file first, then against its normalized form, so either spelling works. Both options apply to an explicit headers array too, and the columns the header check reports are the renamed ones.

Missing columns

A file whose header is wrong fails on its first row, once per field, which reads like a data problem when it is a file problem. checkHeaders looks at the header before the first row is validated:

try {
    await pipeline(createReadStream('employees.csv'), createCsvValidator(Employee), sink);
} catch (error) {
    if (error instanceof MissingColumnsError)
        console.error(
            `missing: ${error.missing.join(', ')} — file has ${error.columns.join(', ')}`
        );
}

Required means the schema cannot do without it: a field that accepts undefined — .optional(), .nullish(), .default() — is not a required column. The check needs an object schema to read; anything else (a z.record, a schema behind .refine()) is left alone. Pass an explicit list to check those, or false to turn it off.

MissingColumnsError stops the stream whatever onInvalidRow says — it is a file-level problem, not a row-level one. The header is checked even when no row ever follows it, so a file that is only a wrong header still fails rather than quietly yielding nothing.

Unknown columns

The mirror of the check above: a file whose email column is spelled emial has every required column missing and one column nobody asked for, and the row errors blame the data. unknownColumns looks at that half too:

createCsvValidator(Employee, { unknownColumns: 'error' });

| Value | Meaning | | ---------- | ------------------------------------------------------------------ | | 'ignore' | Default. Extra columns are the schema's business, not the stream's | | 'warn' | One console.warn naming them, then the file runs as usual | | 'error' | An UnknownColumnsError before the first row, like a missing one |

UnknownColumnsError carries unknown (the columns the schema does not define) and columns (everything in the file). Like MissingColumnsError, it stops the stream whatever onInvalidRow says, and it is measured after normalizeHeaders and columnAliases have had their say — so a renamed column is a known one. checkHeaders: false turns this off along with the missing-column check, and an object schema is what both read: a field that is .optional() is known, just not required.

Did you mean

Both header errors name the closest column on the other side, so the message points at the typo instead of describing it:

CSV is missing column email — the file has name, emial — did you mean "emial" for "email"?

The same pairs are on the error as suggestions, a { wanted: found } map, for a UI that wants to offer the fix rather than print it. Matching folds case and counts a swap of two neighbouring letters as one edit; the budget grows with the length of the name, so a short column is not matched to an unrelated short column.

Structural errors

A record csv-parse cannot read — the wrong number of fields, an unterminated quote — is a parse error, not a validation error, and by default it destroys the stream. skipRecordsWithError sends it down the same channel as an invalid row instead:

const rows = createCsvValidator(Employee, {
    skipRecordsWithError: true,
    onInvalidRow: 'collect',
});

await pipeline(createReadStream('employees.csv'), rows, sink);

for (const error of rows.errors) console.log(error.line, error.name);

Malformed records then arrive as RowParseError through errors, onRowError and 'invalid-row', count against maxErrors alongside the invalid rows, and keep their place in the file: the line numbers of the rows behind them stay right. Under onInvalidRow: 'error' they still stop the stream, just with a positioned RowParseError instead of the parser's own error.

relaxColumnCount is the other half of the story: it lets ragged rows reach the schema instead of failing at all.

Rows the schema erases

A schema is free to output undefined for a row — .transform(row => (row.draft ? undefined : row)) — and the stream carries it through, since undefined is an ordinary object-mode value. Read it with for await, which walks values, rather than a while ((row = stream.read()) != null) loop, which cannot tell that value from a drained buffer. null is the one output that cannot be streamed at all: it ends a Node object-mode stream, so it fails the stream with a TypeError instead.

Row metadata

withMeta wraps every row with where it came from, which is what a source_line column in the database wants:

const rows = createCsvValidator(Employee, { withMeta: true });

for await (const { row, line, record } of rows) await save({ ...row, sourceLine: line });

line is the physical line the record starts on and record its index among the data rows — the same two numbers a RowValidationError carries, so a row and the error next to it agree. It works on the one-shot helpers, through batched(), and on the web build, and the row type follows: CsvRow<Employee> instead of Employee.

Wrapping also gets a schema whose output is null through a Node stream, since what is pushed is the wrapper, not the row.

Progress

A million-row import runs for minutes with nothing to show for it. Every stream carries its own counters, and hands them out on request:

const rows = createCsvValidator(Employee, { onInvalidRow: 'collect' });

setInterval(() => {
    const { bytes, records, valid, invalid } = rows.stats;

    console.log(`${records} records, ${valid} ok, ${invalid} rejected, ${bytes} bytes`);
}, 1000);

| Field | Meaning | | --------- | ------------------------------------------------------------- | | bytes | Bytes csv-parse has read, for a percentage against size | | records | Records it has produced | | valid | Rows the schema accepted | | invalid | Rows it refused, plus the malformed records that were skipped | | dropped | Errors keepErrors had to let go of |

stats is a fresh object each time, so it is safe to keep. onProgress pushes the same thing instead of polling:

const size = (await stat('employees.csv')).size;

createCsvValidator(Employee, {
    onProgress: ({ bytes, records }) => bar.update(bytes / size, { records }),
    progressEveryRecords: 5000,
});

It is called every progressEveryRecords records (default 1000), every progressEveryBytes bytes if you name one, and once more at the end of the stream with the final counts — so a bar always reaches the end. A file with no records never calls it at all. Both the Web Streams build and parseCsv / parseCsvFile carry the same counters; the one-shot helpers return them as stats on the result.

Batches

Most bulk inserts want arrays, not rows:

import { batched, createCsvValidator } from 'csv-zod-stream';

await pipeline(
    createReadStream('employees.csv'),
    createCsvValidator(Employee),
    batched<Employee>(500),
    async function (groups) {
        for await (const employees of groups) await db.insertMany(employees);
    }
);

The last batch is whatever is left over. Backpressure carries through it unchanged. csv-zod-stream/web exports a TransformStream of the same name.

One-shot parsing

When the whole file fits in memory and you want both halves at once:

import { parseCsv, parseCsvFile } from 'csv-zod-stream';

const { rows, errors, droppedErrors, stats } = await parseCsvFile('employees.csv', Employee);
const fromText = await parseCsv('id,name\n1,Ada\n', Employee);

Both collect invalid rows rather than throwing on the first one — pass onInvalidRow to change that — and take the same options as the stream. csv-zod-stream/web exports parseCsv for a string or Uint8Array.

Delimiters

delimiter takes any separator string, or 'auto':

createCsvValidator(Employee, { delimiter: '\t' });
createCsvValidator(Employee, { delimiter: 'auto' });

'auto' buffers up to five lines that are neither blank nor comments, skipping anything inside quotes, and then looks for the candidate — ,, ;, \t or | — that occurs the same number of times on every one of them. That is what a real delimiter does; a stray separator inside a header field does not, so Name, full;age over a;1 still reads as ;. If nothing is consistent it falls back to sheer frequency, and then to ,. It honours quote and escape. It is still a heuristic — name the delimiter when you know it.

Encodings

Not every export is UTF-8. encoding decodes the bytes on the way in, using the runtime's TextDecoder, so anything it knows works — windows-1251, windows-1252, latin1, koi8-r, utf-16le:

await parseCsvFile('employees.csv', Employee, { encoding: 'windows-1251' });

Decoding is streaming: a character split across two chunks is put back together, not mangled. It happens before the delimiter sniffer and before csv-parse, so everything downstream keeps seeing UTF-8. A leading BOM is stripped unless bom: false. When the input to parseCsv is already a string the option is ignored — there is nothing left to decode.

Web Streams

csv-zod-stream/web is the same validator over TransformStream, with no Node stream machinery in the bundle:

import { createCsvValidator } from 'csv-zod-stream/web';

const response = await fetch('https://example.com/employees.csv');

for await (const employee of response.body.pipeThrough(createCsvValidator(Employee))) {
    await save(employee);
}

Backpressure comes from pipeThrough itself — nothing is pulled out of the parser until the consumer asks for the next row.

There is no EventEmitter here, so pass onRowError instead of listening for 'invalid-row'; .errors and .droppedErrors work the same as on the Node build.

const validator = createCsvValidator(Employee, {
    onInvalidRow: 'collect',
    onRowError: error => log.warn(error.line, error.message),
});

This build reaches Web Streams through csv-parse/stream, which imports them from node:stream/web. Node, Deno, Bun and edge runtimes with Node compatibility resolve that; a plain browser bundle needs your bundler to alias node:stream/web to the platform globals.

Options

| Option | Default | Meaning | | ---------------------- | ---------- | ---------------------------------------------------------------------------- | | delimiter | ',' | Field separator, or 'auto' to sniff it | | headers | true | true reads names from the first row; an array names a headerless file | | checkHeaders | true | Fail fast on missing columns; false to skip, or an explicit column list | | unknownColumns | 'ignore' | What to do about columns the schema does not define: 'warn' | 'error' | | normalizeHeaders | — | Fold the header row: 'trim', 'lower', 'snake', 'camel' or a function | | columnAliases | — | Rename file columns onto schema fields | | encoding | 'utf-8' | Decode the bytes with this encoding before parsing | | emptyAs | 'keep' | Turn blank cells into undefined or null before validating | | async | false | Validate with safeParseAsync, one row at a time | | withMeta | false | Emit { row, line, record } instead of the row | | onInvalidRow | 'error' | 'error' | 'skip' | 'collect' | | maxErrors | Infinity | Invalid rows tolerated under 'skip' / 'collect' | | keepErrors | 1000 | Invalid rows kept in memory before the oldest is dropped | | onRowError | — | Called for each invalid row under 'skip' / 'collect' | | onProgress | — | Called with stats on the interval below, and once at the end | | progressEveryRecords | 1000 | Records between onProgress calls | | progressEveryBytes | — | Bytes between onProgress calls, if you would rather pace by size | | skipRecordsWithError | false | Route malformed records through the invalid-row channel | | bom | true | Strip a leading UTF-8 BOM | | skipEmptyLines | true | Ignore blank lines rather than treating them as records | | relaxColumnCount | false | Let ragged rows through so the schema judges them | | trim | false | Trim whitespace around unquoted fields | | quote | '"' | Quote character | | escape | '"' | Escape character inside quoted fields | | comment | — | Treat this character and the rest of its line as a comment | | parse | — | Raw csv-parse options |

parse — the escape hatch

Anything csv-parse supports and this package does not name is reachable through parse:

createCsvValidator(Employee, {
    parse: { from_line: 3, to_line: 1000, max_record_size: 1_000_000, record_delimiter: '\u2028' },
});

The options above win over their parse equivalents when you set them, and fill in from parse when you do not. Four are the library's own and cannot be taken over: delimiter, columns, info and raw — the line numbers are built out of the last two.

Bad values are refused where you pass them, not where they would eventually misbehave: an unknown emptyAs, onInvalidRow or normalizeHeaders, a negative maxErrors or keepErrors, an empty delimiter and headers that are not strings all throw a RangeError out of createCsvValidator itself.

comment is csv-parse's: the character opens a comment wherever it appears outside a quoted field, so a,1 # note parses age as 1, not 1 # note. Pass parse: { comment_no_infix: true } to have it count only at the start of a line.

Not yet

CSV generation from objects and a browser File helper are both out of scope for now.

License

MIT © Pavel Lazarchuk