阅读中文版本

Build a CSV Import Validation Pipeline

A complete CSV import pipeline: streaming parse, row-level validation, error collection, transactional commit, and partial-success tests.

csv-import-pipeline/
├── src/parse.ts
├── src/validate.ts
├── src/stage.ts
├── src/commit.ts
├── tests/import.test.ts
└── package.json

import { parse } from 'csv-parse';
input.pipe(parse({ columns: true, skip_empty_lines: true })).on('data', handleRow);

if (!isEmail(row.email)) errors.push({ line, field: 'email', value: row.email });

if (batch.length >= 500) { await insertBatch(batch); batch = []; }

if (errors.length > maxErrors) return rollback(batchId);

res.json({ batchId, accepted, rejected: errors.length, errors: errors.slice(0, 100) });

Project overview

Bulk imports fail in the worst possible way: halfway through, leaving a partially loaded table and no clear answer about which rows made it. This build validates every row, collects all errors with line numbers, and commits only when the batch as a whole is acceptable.

You will stream a large CSV instead of buffering it, validate each row against a schema while accumulating typed errors, stage accepted rows, and commit them in one transaction only if the error count is within your configured threshold.

The design treats a batch as all-or-nothing by default, because a half-applied import is far harder to reason about than a rejected one. A threshold exists for the case where a few bad rows are tolerable, and even then the decision is explicit.

The report matters as much as the import. Giving the operator the exact line numbers and reasons turns a failed upload into a fixable CSV rather than a mystery.

Each phase leaves a usable importer: parsing, then validation, then staging, then the commit policy, then the error report, so you can run it safely at every step.

The importer exists because the alternative is a spreadsheet, a hand-written loader, and a production table that drifts from the source of truth. Making validation explicit is what turns an import into a repeatable, reviewable operation.

What you will build

The pipeline provides five capabilities, each verified by feeding it CSVs with known problems:

• Streaming parse
• Row validation
• Staging
• Commit policy
• Error report

A streaming parser keeps memory flat; a row validator reports typed errors with line numbers; staging isolates the load; a commit policy decides all-or-nothing; and the error report tells the operator exactly what to fix.

Nothing here loads the whole file or trusts a header blindly. The importer is built for the files that actually arrive: large, slightly malformed, and full of the outliers manual testing never sees.

Architecture and data model

A streaming parser feeds rows into a validator that accumulates typed errors. Accepted rows are staged, and a final transaction inserts them only if the error count is within the configured threshold, otherwise the batch is rejected with the error report.

A streaming parser emits one record at a time into a validator that accumulates typed errors rather than throwing on the first bad row. Accepted rows are written to a staging table, and a final step either promotes them or discards the batch.

Errors carry a line number, a column, and a code, so the report can point at the exact cell. Aggregating errors instead of stopping at the first lets an operator fix a file once rather than iteratively.

The staging table mirrors the target schema but stays separate, so a rejected batch leaves the live table untouched. Because staging is keyed by batch id, a re-import can be reconciled with the previous attempt rather than silently duplicating rows.

The commit threshold is explicit configuration: zero errors means strict and a positive number tolerates a known proportion of bad rows. Making the number visible prevents the drift where an import quietly accepts more garbage than intended.

The parser and the validator are separate so that a format change and a schema change are independent events. Coupling them is convenient until the day a column is renamed and the parsing rules silently stop matching.

Failed rows are retained in the error accumulator together with their raw line, so the report can quote the offending cell without re-reading the file. Keeping the evidence beside the error is what makes a report actionable rather than merely alarming.

Prerequisites and dependencies

You need a database with a staging area for the target table and a schema that describes the incoming rows.

• Node.js 20 LTS or later
• PostgreSQL 15+
• A sample CSV with deliberate errors

The dependencies keep parsing, validation, and storage in one process, which is enough for files that fit a single machine:

• csv-parse (^5.5.6) — Streaming CSV parser
• pg (^8.12.0) — Staging and commit

pnpm init
pnpm add csv-parse pg
pnpm add -D typescript tsx vitest

Pin the schema and the parser version together, since a change to how quotes or encodings are interpreted alters which rows are valid. Keep the commit threshold in configuration and treat a change to it as a reviewed event.

Directory structure

Separate parsing, validation, staging, the commit decision, and the error report so each can be exercised with a small fixture:

csv-import-pipeline/
├── src/parse.ts
├── src/validate.ts
├── src/stage.ts
├── src/commit.ts
├── tests/import.test.ts
└── package.json

Phase 1: Streaming parse

Feed the stream through the parser and process row events.

import { parse } from 'csv-parse';
input.pipe(parse({ columns: true, skip_empty_lines: true })).on('data', handleRow);

Reading the whole file into memory fails on large imports.

Stream the file and emit records as they parse, holding only one row and the error accumulator in memory. A parser that reads the whole file first will pass tests and fail on the first real upload.

Phase 2: Row validation

Coerce and validate each field, collecting errors with the line number.

if (!isEmail(row.email)) errors.push({ line, field: 'email', value: row.email });

Users need to know exactly which rows failed and why.

Validate each row against the schema and record every failure with its line number and column instead of aborting. Collecting all errors is what turns a frustrating iterative fix into a single actionable report.

Phase 3: Staging

Buffer accepted rows in bounded batches.

if (batch.length >= 500) { await insertBatch(batch); batch = []; }

Batched inserts are far faster than one insert per row.

Write validated rows to a staging table keyed by the batch id. Staging is the boundary that makes the commit reversible, because nothing touches the target until the batch is judged acceptable.

Phase 4: Commit policy

Commit only if errors are within the accepted threshold.

if (errors.length > maxErrors) return rollback(batchId);

A partial import can silently lose money or data.

Decide the batch outcome by comparing the error count to the threshold and commit in one transaction. The threshold is the only place that encodes how strict the import is, so it should be explicit and tested.

Phase 5: Error report

Return a downloadable report of every rejected row.

res.json({ batchId, accepted, rejected: errors.length, errors: errors.slice(0, 100) });

Ops can fix the source and re-import precisely.

Emit a report that names each failing row and reason, and keep it retrievable by batch id. An operator who can see the first twenty problems fixes the file; one who sees a single generic failure gives up.

Testing

Import a CSV with valid and invalid rows and assert the accepted count, that the error list references the right line numbers, and that an over-threshold batch commits nothing.

Import a CSV mixing valid and invalid rows and assert the accepted count and that each error references the correct line number. Off-by-one line numbers are the classic bug here, so pin them with a fixture that has a header.

Feed a batch whose error count exceeds the threshold and assert that nothing was committed and the target table is unchanged, which is the all-or-nothing guarantee stated as a test.

Re-import the same batch id and assert the result is reconciled rather than duplicated, which proves staging behaves as a boundary and not as a second target.

Test a file with a quoted field containing a comma and a newline, because that is the case a naive split-on-comma parser gets wrong and the one real CSVs are full of.

Pin the error codes as well as the line numbers, because an operator tooling against the report depends on the codes staying stable across releases.

pnpm vitest run

Deployment and operations

Run imports against a staging table and promote atomically. Track the batch id so a re-import can be reconciled with the previous attempt.

Run imports against a staging table and promote atomically, so a failed import costs storage but never corrupts the live table. Track the batch id so a retry can be reconciled with the previous attempt rather than creating duplicates.

Import from a job separate from request handling, because a large file will otherwise hold a request open long past any sensible timeout.

Keep a copy of the rejected report for at least one cycle, since an operator often needs the exact reasons after they have fixed and re-uploaded the file.

Set a maximum file size and row count and reject beyond it before parsing, because an unbounded upload is a denial-of-service vector as much as a data-quality one.

Cap the staging table with the same retention as the report, so a rejected batch does not become a permanent archive of raw uploads that nobody owns.

Security considerations

Validate and size-cap uploads before parsing, reject unexpected columns, and never execute formulas from CSV cells in downstream spreadsheets.

Never interpolate a CSV-derived value directly into SQL; bind parameters for every column, since a spreadsheet is untrusted input like any other.

Escape formula-leading characters when exporting or echoing a rejected row back to a spreadsheet UI, because a cell beginning with a formula character is an injection into the operator’s tool.

Restrict who can trigger an import and keep the staging area separate from production tables, so a malformed batch cannot reach live data even if validation is bypassed.

Treat the uploaded file as sensitive data and purge it under the same policy as the rows it contains, because a CSV is often a copy of a production table.

Log only counts and the batch id at the info level and keep row contents out of logs, because a rejected row is often a copy of sensitive production data.

Observability

Track rows per second, error rate by field, and batch rejection rate. Alert when the error rate jumps, which indicates a source schema change.

Track rows parsed, rows accepted, rows rejected, and the accept ratio per batch. A slow decline in the ratio is a data-quality trend that is invisible in any single import.

Chart the time from upload to commit, because a growing duration usually means the parser is buffering more than it should.

Log the batch id with every outcome so support can join a complaint to the exact import, and alert when a batch is rejected so the operator hears it from the system rather than the customer.

Measure staging table growth, since forgotten rejected batches accumulate silently and consume storage until a cleanup job exists.

Track the distinct error codes per import, since a single recurring code across batches points at a format assumption that needs fixing upstream rather than repeated cleanup.

Track the size distribution of imports, because an import that suddenly doubles in size usually means an upstream export changed shape as well as volume.

Troubleshooting

Import is slow. Rows were inserted one at a time; batch inserts within a transaction.

Line numbers are off. Header rows or blank lines shifted the counter; count from the parser row context.

When a valid-looking row is rejected, inspect the raw bytes of that line rather than the rendered view; a smart quote or a byte-order mark is the usual culprit, and both hide in a text editor.

If a batch commits fewer rows than were parsed, check the threshold and the error report together; the difference is the rows that were rejected, and the report names each one.

Limitations and next steps

• No type inference beyond configured columns.
• Single-file imports only.
• The threshold is global, not per column.

Validation is schema-based rather than semantic: types and shapes are checked, but references to rows in other tables are not resolved during import.

The pipeline is single-process, so very large files are limited by one machine’s throughput. Parallel chunked import is a reasonable extension but introduces ordering requirements this build does not handle.

There is no automatic mapping UI for renaming columns; the incoming schema is fixed and documented. A mapping layer is the next step for files from varied sources.

Cross-batch deduplication is out of scope, so importing the same file twice requires an operator to use the batch id rather than relying on the pipeline to detect overlap.

The importer handles a single file per batch and does not merge multiple files into one logical load, so a sharded export needs concatenation before upload.

There is no incremental resume: a rejected batch must be fixed and re-imported from the start, since staging is keyed to a whole batch rather than to a byte offset.