CSV & JSON Transform Pipelines

Data-heavy front ends routinely ingest multi-megabyte CSV exports and JSON payloads, and parsing them on the main thread is the single most common source of a frozen tab in an analytics UI. This guide is part of the High-Performance Computation Patterns reference and walks the whole extract-transform-load path end to end: slicing a File, transferring bytes into a worker, parsing CSV with a state machine that survives chunk boundaries, transforming and validating rows, choosing how results cross the thread boundary, and proving with real measurements that the main thread never stalls.

The Jank You Are Trying to Remove

A 5 MB CSV file parsed synchronously with a naive split('\n').map(parseLine) loop blocks the main thread for 80-300 ms on mid-range hardware. During that window the browser cannot respond to scroll, cannot repaint an animation, and queues every keystroke. At 60 fps each frame has a 16.7 ms budget, so a 200 ms parse burns twelve consecutive frames. Chrome’s Performance panel shows it as one unbroken long task; users describe it as “the page died for a second”.

The symptom scales badly in three directions at once. Parse time grows linearly with file size. Peak memory grows with the whole result set if you accumulate rows before rendering. And garbage-collection pauses grow with the number of small objects allocated, because a million-row CSV mapped to plain objects produces a million short-lived allocations that the collector must trace.

The fix is not a faster parser. It is a different shape of work:

1. Move the parser off the main thread so its cost never touches the frame budget.
2. Stream the input in fixed-size chunks so peak memory stays flat regardless of file size.
3. Return results incrementally in batches so the UI paints rows as they arrive rather than after the last byte.

Each of those three decisions has a cost, and most of this page is about paying those costs deliberately rather than discovering them in production.

Prerequisites

Before implementing the pipeline, confirm you have:

• A measured baseline. A performance.now() delta around your existing parse, on your target hardware, with a representative file. If it is under 16 ms, a worker adds complexity for nothing.
• The Worker constructor and message API. Familiarity with postMessage, onmessage, and the fact that the worker has no DOM access. If thread lifecycle is new to you, start with Main Thread vs Worker Thread Lifecycle.
• Typed arrays. ArrayBuffer, Uint8Array and Float64Array, plus what “detached” means after a transfer — covered in Transferable Objects & Zero-Copy.
• A bundler that understands module workers. Vite, webpack 5, Rollup and esbuild all support new Worker(new URL('./worker.ts', import.meta.url), { type: 'module' }); see Bundling Module Workers with Vite and Webpack for the per-bundler configuration.
• COOP/COEP headers only if you plan to use shared memory. The default pipeline here does not need them. The shared-memory variant in the data-transfer section does, and it is opt-in.

Thread Boundaries for a Transform Pipeline

The split of responsibilities is not negotiable if you want a steady frame rate. The main thread owns everything that touches the DOM and the user; the worker owns everything that touches bytes.

Thread safety comes from having no shared mutable state at all. Everything crossing the boundary is either structured-cloned or transferred, and once a buffer is transferred the sender genuinely cannot read it any more. That constraint is what makes the design safe: there is no lock to forget.

One nuance worth stating early — the main thread still does some work per chunk. It slices the file, awaits the arrayBuffer() read, and calls postMessage. On a 100 MB file at 1 MB chunks that is 100 slices, each costing a fraction of a millisecond. That is fine. What is not fine is await file.arrayBuffer() on the whole file, which allocates the entire thing on the main thread’s heap before you have parsed a single row.

Step-by-Step Implementation

Step 1 — Fix the message contract before writing any parser code

Untyped postMessage payloads are where worker pipelines rot. Define the full message union once, in a shared module imported by both sides, so a renamed field is a compile error rather than a silent undefined at runtime.

// pipeline-protocol.ts — imported by BOTH main thread and worker
export type ToWorker =
  | { type: 'INIT'; headers?: string[]; delimiter?: string; batchSize?: number }
  | { type: 'CHUNK'; seq: number; buffer: ArrayBuffer }
  | { type: 'CHUNK_END' }
  | { type: 'ABORT' };

export type FromWorker =
  | { type: 'READY' }
  | { type: 'PROGRESS'; seq: number; rowsParsed: number; bytesParsed: number }
  | { type: 'BATCH'; rows: Record<string, string | number>[] }
  | { type: 'COLUMNS'; names: string[]; buffers: ArrayBuffer[]; rowCount: number }
  | { type: 'ERROR'; error: SerializedError; seq?: number; fatal: boolean }
  | { type: 'COMPLETE'; rowsParsed: number; rowsRejected: number; elapsedMs: number };

export interface SerializedError {
  name: string;
  message: string;
  stack?: string;
  /** Row index that triggered the failure, when the error is row-scoped. */
  row?: number;
}

The main-thread controller then becomes a thin, fully typed wrapper. Note that it exposes a promise for completion but delivers batches through a callback, because waiting for the promise defeats the point of streaming.

// pipeline-controller.ts
import type { FromWorker, ToWorker } from './pipeline-protocol';

export interface PipelineOptions {
  chunkSize?: number;
  onBatch: (rows: Record<string, string | number>[]) => void;
  onProgress?: (bytesParsed: number, totalBytes: number) => void;
}

export function runPipeline(file: File, opts: PipelineOptions): Promise<number> {
  const { chunkSize = 1024 * 1024 } = opts;
  const worker = new Worker(new URL('./transform.worker.ts', import.meta.url), {
    type: 'module',
  });

  return new Promise<number>((resolve, reject) => {
    worker.onmessage = (e: MessageEvent<FromWorker>) => {
      const msg = e.data;
      switch (msg.type) {
        case 'BATCH':
          opts.onBatch(msg.rows);
          break;
        case 'PROGRESS':
          opts.onProgress?.(msg.bytesParsed, file.size);
          break;
        case 'ERROR':
          if (msg.fatal) { worker.terminate(); reject(Object.assign(new Error(msg.error.message), msg.error)); }
          else console.warn(`[pipeline] row ${msg.error.row}: ${msg.error.message}`);
          break;
        case 'COMPLETE':
          worker.terminate();          // always terminate on BOTH paths
          resolve(msg.rowsParsed);
          break;
      }
    };

    // Worker-level failures (a syntax error in the worker module, an uncaught throw)
    // never reach onmessage — they surface here instead.
    worker.onerror = (e) => { worker.terminate(); reject(new Error(`worker: ${e.message}`)); };
    worker.onmessageerror = () => { worker.terminate(); reject(new Error('worker: payload was not cloneable')); };

    const post = (msg: ToWorker, transfer: Transferable[] = []) => worker.postMessage(msg, transfer);
    post({ type: 'INIT', batchSize: 5000 });
    void streamFile(file, chunkSize, post);
  });
}

Step 2 — Stream the file in fixed-size slices

File.slice() returns a lazy Blob view; no bytes are read until you call arrayBuffer() on it. That makes peak main-thread memory one chunk, not one file.

// stream-file.ts
import type { ToWorker } from './pipeline-protocol';

export async function streamFile(
  file: File,
  chunkSize: number,
  post: (msg: ToWorker, transfer?: Transferable[]) => void,
): Promise<void> {
  let offset = 0;
  let seq = 0;

  while (offset < file.size) {
    const slice = file.slice(offset, Math.min(offset + chunkSize, file.size));
    const buffer = await slice.arrayBuffer();     // only this chunk is resident
    post({ type: 'CHUNK', seq: seq++, buffer }, [buffer]);  // ownership moves; buffer is now detached
    offset += chunkSize;
  }

  post({ type: 'CHUNK_END' });
}

The seq counter is not decoration. postMessage guarantees ordering into a single worker, so with one worker the sequence numbers only serve diagnostics — but the moment you fan chunks across several workers, ordering is gone and the receiver must reassemble by seq. Tag from day one so the fan-out refactor is not a rewrite.

If you prefer the streams API, file.stream().getReader() gives you Uint8Array chunks sized by the browser (typically 64 KB) and removes the manual offset arithmetic. The trade-off is chunk size: 64 KB chunks mean 16× more postMessage calls per megabyte, and at that granularity per-message overhead starts to show. Slicing at 1 MB keeps message count low while keeping each serialization step under 2 ms.

Step 3 — Decode and parse with a chunk-boundary-safe state machine

This is the step most implementations get wrong. A 1 MB slice knows nothing about record boundaries: it will cut a line in half, and roughly one in every few thousand chunks will cut a multi-byte UTF-8 character in half. Parsing each chunk independently corrupts data at every boundary.

Two mechanisms solve it. TextDecoder in streaming mode buffers a partial code-point sequence internally and emits it once the next chunk supplies the missing bytes. And the parser itself is written as a state machine that keeps its field buffer, current row and in-quotes flag between calls.

// csv-state-machine.js — worker-side, RFC 4180 semantics
export function createCsvParser({ delimiter = ',', onRow }) {
  let field = '';
  let row = [];
  let inQuotes = false;
  let quotePending = false;   // chunk ended on a '"' while inside quotes

  const endField = () => { row.push(field); field = ''; };
  const endRow = () => { endField(); const out = row; row = []; onRow(out); };

  return {
    /** Feed one decoded chunk. Safe to call with a record split across calls. */
    write(text) {
      let i = 0;

      // Resolve a quote that straddled the previous chunk boundary:
      // '""' is an escaped quote, anything else closed the quoted field.
      if (quotePending) {
        quotePending = false;
        if (text[0] === '"') { field += '"'; i = 1; }
        else { inQuotes = false; }
      }

      for (; i < text.length; i++) {
        const ch = text[i];

        if (inQuotes) {
          if (ch !== '"') { field += ch; continue; }
          if (i === text.length - 1) { quotePending = true; continue; }  // decide next chunk
          if (text[i + 1] === '"') { field += '"'; i++; }                // escaped quote
          else { inQuotes = false; }
          continue;
        }

        if (ch === '"' && field === '') { inQuotes = true; continue; }
        if (ch === delimiter) { endField(); continue; }
        if (ch === '\r') continue;                 // CRLF and LF both terminate a record
        if (ch === '\n') { endRow(); continue; }
        field += ch;
      }
    },

    /** Flush the trailing record on files with no final newline. */
    end() {
      if (quotePending) { quotePending = false; inQuotes = false; }
      if (field !== '' || row.length > 0) endRow();
    },
  };
}

Because the state machine is a character loop rather than a regular expression, quoted fields containing delimiters, embedded newlines and escaped double-quotes all fall out for free — and there is no catastrophic-backtracking risk on adversarial input. On a 2019-class laptop it sustains roughly 40-70 MB/s of CSV, which means the parse itself is rarely the bottleneck; message overhead and object allocation usually are.

Wiring it into the worker:

// transform.worker.js
import { createCsvParser } from './csv-state-machine.js';

const decoder = new TextDecoder('utf-8');   // reused so streaming state survives chunks
let headers = null;
let parser = null;
let batch = [];
let batchSize = 5000;
let rowsParsed = 0;
let bytesParsed = 0;
let startedAt = 0;

function handleRow(values) {
  if (headers === null) { headers = values.map((h) => h.trim()); return; }  // first record is the header
  const record = {};
  for (let i = 0; i < headers.length; i++) record[headers[i]] = values[i] ?? '';
  batch.push(record);
  rowsParsed++;
  if (batch.length >= batchSize) flush();
}

function flush() {
  if (batch.length === 0) return;
  self.postMessage({ type: 'BATCH', rows: batch });
  batch = [];                                  // drop the reference so the batch can be collected
}

self.onmessage = (e) => {
  const msg = e.data;

  if (msg.type === 'INIT') {
    batchSize = msg.batchSize ?? 5000;
    parser = createCsvParser({ delimiter: msg.delimiter ?? ',', onRow: handleRow });
    startedAt = performance.now();
    self.postMessage({ type: 'READY' });
    return;
  }

  if (msg.type === 'CHUNK') {
    bytesParsed += msg.buffer.byteLength;
    // stream: true keeps a partial multi-byte sequence buffered for the next chunk
    parser.write(decoder.decode(msg.buffer, { stream: true }));
    self.postMessage({ type: 'PROGRESS', seq: msg.seq, rowsParsed, bytesParsed });
    return;
  }

  if (msg.type === 'CHUNK_END') {
    parser.write(decoder.decode());   // flush any buffered bytes
    parser.end();
    flush();
    self.postMessage({
      type: 'COMPLETE', rowsParsed, rowsRejected: 0,
      elapsedMs: performance.now() - startedAt,
    });
  }
};

Step 4 — Transform rows through an async generator

Blocking Array.map() chains over a million rows build a second full-size array in memory before the first row is used. An async generator pipeline yields rows one at a time, keeps peak memory at the batch size, and gives you a natural place to await an async enrichment step. The refactor patterns — and the leak traps that come with them — are covered in Migrating Synchronous Loops to Web Workers Safely.

// transform-stage.js — worker-side
async function* transformRows(rows, rules) {
  for (let i = 0; i < rows.length; i++) {
    try {
      yield rules(rows[i], i);
    } catch (err) {
      // A bad row is data, not a crash: report it and keep the stream alive.
      self.postMessage({
        type: 'ERROR', fatal: false,
        error: { name: err.name, message: err.message, row: i },
      });
    }
  }
}

export async function processBatch(rows, rules, batchSize = 5000) {
  const out = [];
  for await (const transformed of transformRows(rows, rules)) {
    out.push(transformed);
    if (out.length >= batchSize) self.postMessage({ type: 'BATCH', rows: out.splice(0) });
  }
  if (out.length > 0) self.postMessage({ type: 'BATCH', rows: out });
}

The try sits inside the generator, per row. Wrapping the whole for await loop instead would abandon every remaining row in the batch on the first malformed record — the difference between “17 rows rejected” and “import failed”.

Step 5 — Validate against a schema without halting the stream

Validation belongs in the worker, inline with the transform, so invalid rows never occupy main-thread memory. Keep it declarative and keep it cheap: a per-row closure compiled once beats re-reading a schema object for every field.

// validate.js — worker-side
export function compileValidator(schema) {
  const required = schema.required ?? [];
  const coercions = Object.entries(schema.types ?? {});

  return function validate(record, index) {
    for (const key of required) {
      if (record[key] === undefined || record[key] === '') {
        return { ok: false, row: index, reason: `missing required field "${key}"` };
      }
    }
    for (const [key, type] of coercions) {
      const raw = record[key];
      if (raw === undefined || raw === '') continue;
      if (type === 'number') {
        const n = Number(raw);
        if (Number.isNaN(n)) return { ok: false, row: index, reason: `"${key}" is not numeric` };
        record[key] = n;                       // coerce in place — the row is worker-owned
      } else if (type === 'date') {
        const t = Date.parse(raw);
        if (Number.isNaN(t)) return { ok: false, row: index, reason: `"${key}" is not a date` };
        record[key] = t;                       // epoch ms survives structured clone cheaply
      }
    }
    return { ok: true };
  };
}

Coercing dates to epoch milliseconds instead of Date objects is a deliberate serialization choice: Date instances are structured-cloneable, but each one is a separate heap object to allocate on both sides, whereas a number is eight bytes that can later live in a Float64Array column. The same logic applies to numbers parsed once here rather than repeatedly at render time — a principle explored further in Data Parsing & Serialization.

Report rejections in aggregate rather than one message per bad row. A file with 50,000 malformed records would otherwise generate 50,000 messages and reproduce the very bottleneck you removed.

Choosing the Data-Transfer Mechanism

Getting bytes into the worker is settled — transfer the ArrayBuffer, always. Getting results out is the real decision, and it has three answers with very different cost curves.

Structured clone of row objects. The default. postMessage({ rows }) deep-copies the array; the algorithm walks every object, every key and every string, hashing references to detect cycles. Cost tracks object and property count, not byte count: roughly 0.5-1 ms per 1,000 rows of a dozen short fields. Up to about 20,000 rows per message this is invisible and the code is trivial. The mechanics are detailed in Step-by-Step Guide to the Structured Clone Algorithm.

Columnar typed arrays plus transfer. For large numeric result sets, stop shipping objects. Encode each numeric column into its own Float64Array and transfer the underlying buffers: the clone step disappears entirely because only ownership moves.

// columnar.js — worker-side result encoding
export function encodeColumns(rows, numericFields) {
  const buffers = [];
  const names = [];

  for (const field of numericFields) {
    const col = new Float64Array(rows.length);
    for (let i = 0; i < rows.length; i++) col[i] = rows[i][field] ?? NaN;
    names.push(field);
    buffers.push(col.buffer);
  }

  // Transfer every column buffer: zero-copy, and the worker's copies detach.
  self.postMessage({ type: 'COLUMNS', names, buffers, rowCount: rows.length }, buffers);
}

On the main thread each buffer is rewrapped with new Float64Array(buffer) and fed straight into a chart library or a virtualized table. A 100,000-row × 4-column result is 3.2 MB of contiguous memory that crosses the boundary in well under a millisecond, versus tens of milliseconds to clone 100,000 objects. This is the same reasoning applied to JSON in Streaming JSON Parsing with Transferable Chunks, and the profiling technique for confirming the win lives in postMessage Bottleneck Analysis.

SharedArrayBuffer with Atomics. Warranted only for a continuous pipeline — several parser workers feeding one aggregator, or a live feed where results are produced faster than the consumer drains them. A shared ring buffer removes even the ownership hop and lets the consumer read while the producer writes, coordinated with Atomics.wait/Atomics.notify. The cost is real: the page must be cross-origin isolated, which means serving Cross-Origin-Opener-Policy: same-origin and Cross-Origin-Embedder-Policy: require-corp, and every cross-origin subresource must then send CORP or CORS headers or it fails to load. Feature-detect with crossOriginIsolated and keep a transfer-based fallback. See SharedArrayBuffer & Atomics for the ring-buffer construction.

Verification & Measurement

A pipeline that “feels faster” is not a result. Two measurements settle it: end-to-end elapsed time, and the longest main-thread task during the run.

// measure.ts — main thread
const t0 = performance.now();
performance.mark('pipeline:start');

const rowCount = await runPipeline(file, {
  onBatch: (rows) => table.appendRows(rows),
});

performance.mark('pipeline:end');
performance.measure('pipeline', 'pipeline:start', 'pipeline:end');
console.log(`${rowCount} rows in ${(performance.now() - t0).toFixed(1)} ms`);

Elapsed time alone can hide jank, so watch long tasks at the same time. Anything the observer reports above 50 ms during the run is a task that dropped frames:

new PerformanceObserver((list) => {
  for (const entry of list.getEntries()) {
    console.warn(`long task ${entry.duration.toFixed(1)} ms at ${entry.startTime.toFixed(0)} ms`);
  }
}).observe({ type: 'longtask', buffered: true });

Inside the worker, performance.mark and performance.measure are available and — importantly — their entries appear on the worker’s own track in a Chrome Performance recording. Marking parse, validate and encode separately tells you which stage actually dominates instead of leaving you guessing.

What a healthy 10 MB CSV run looks like:

• Total elapsed: 200-600 ms for numeric transforms on mid-range hardware.
• Longest main-thread task: under 5 ms — the slice-and-post loop, plus whatever your render callback does per batch.
• Long-task warnings: zero.
• Performance panel shape: the Main track is nearly idle with small periodic blips; the Worker track carries one continuous busy band.

A quick smoke-test checklist before you call it done:

• A 1 MB file completes end to end in under 50 ms.
• The Performance panel shows no long task on Main during parsing.
• CHUNK_END always produces exactly one COMPLETE, including for an empty file and a file with no trailing newline.
• A fixture with quoted commas, embedded newlines, "" escapes and a non-ASCII character positioned deliberately across a chunk boundary round-trips byte-identically.
• Validation reports a non-zero rejection count for a known-bad fixture rather than silently passing it.

Failure Modes & Error Handling

Worker errors do not propagate to the main thread’s window.onerror. If you do not wire the handlers, a worker that throws simply goes quiet and your promise never settles.

Uncaught throws inside the worker fire worker.onerror on the main thread with message, filename and lineno — but not the original error object, because Error instances lose their prototype across the boundary. Catch and serialize deliberately instead:

// transform.worker.js — structured error reporting
function reportFatal(err, seq) {
  self.postMessage({
    type: 'ERROR',
    fatal: true,
    seq,
    error: { name: err.name, message: err.message, stack: err.stack },
  });
}

self.addEventListener('error', (e) => { e.preventDefault(); reportFatal(e.error ?? new Error(e.message)); });
self.addEventListener('unhandledrejection', (e) => { e.preventDefault(); reportFatal(e.reason); });
self.addEventListener('messageerror', () => {
  reportFatal(new Error('inbound message could not be deserialized'));
});

unhandledrejection matters more than it looks in this pipeline, because the async generator stages return promises: a rejected await inside a stage produces no error event at all, only an unhandled rejection. The wider patterns for this live in Error Handling & Crash Recovery, and the serialization shape that survives a trip to your telemetry backend is covered in Structured Error Serialization Across Threads.

The specific ways this pipeline breaks:

Detached ArrayBuffer after transfer. Once a buffer is in the transfer list, the sender’s view is detached: byteLength reads 0 and any typed-array access throws. Decode or copy before transferring, and never keep the buffer in a retry cache — cache the file offset and re-slice instead, which is free because File.slice() is lazy.

Chunk boundary corruption. Covered in Step 3, but it deserves a regression test rather than trust: build a fixture where a multi-byte character and a quoted field each straddle a deliberate boundary, then parse it at chunk sizes of 16, 17 and 18 bytes. Off-by-one state-machine bugs surface immediately at tiny chunk sizes and hide completely at 1 MB.

Out-of-order chunks in a worker pool. Ordering holds within one worker only. Fan chunks across a pool and records that span a boundary end up in different workers with no shared parser state. Either keep parsing single-threaded and parallelise only the transform stage, or pre-split the input on record boundaries before dispatch.

Unbounded batch queues. If the worker parses faster than the main thread renders — common, since parsing is 40+ MB/s and DOM insertion is not — BATCH messages pile up in the main thread’s task queue and memory climbs until the tab dies. Apply backpressure: have the main thread acknowledge each batch and have the worker pause parsing once more than two batches are unacknowledged.

Leaked workers. worker.terminate() must run on the success path, the error path and the abort path. A leaked worker holds its parser state, its decoder, and every transferred buffer it ever received; nothing in that graph is collectable while the worker lives.

Browser Compatibility

The core path — Worker, File.slice(), buffer transfer, streaming TextDecoder — has been universally supported for years, so the pipeline itself carries no compatibility risk. The cliffs are all optional: module workers need Firefox 114+ (a classic worker plus importScripts is the fallback), shared memory needs isolation headers, and the Long Tasks API is Chromium-only, so treat it as a development-time instrument rather than production telemetry. CompressionStream is worth using only when result payloads are large and repetitive, and only behind a feature check with a plain postMessage fallback.

For datasets past 50 MB where the JavaScript state machine becomes the limiting factor, the next step is a parser compiled to WebAssembly running in the same worker — the streaming and transfer architecture on this page stays exactly as it is, and only the parse stage changes. That path is covered in WebAssembly in Workers.

Thread safety and memory discipline are what make this pattern hold up under real data. Strict message boundaries, transferred buffers instead of cloned object graphs, a parser that keeps state across chunks, batched results with backpressure, and errors that are serialized rather than swallowed — together these let a browser tab process enterprise-scale files while the interface keeps painting at 60 fps.

Going Further

Whether to write the parser at all is worth deciding deliberately. Papa Parse vs a Custom Worker CSV Parser sets out the RFC 4180 edge cases a hand-written parser has to handle, the allocation costs that decide the throughput comparison, and the detect-and-fall-back arrangement that lets a narrow fast path coexist with a correct general one.