NDJSON 流式校验解析

示例说明

数据管道里坏行是常态，处置策略决定管道韧性：整批失败会因个别脏数据反复阻塞，全放行则污染下游。校验 Transform 把裁决放进流中：行解析失败或过不了 schema 的进入隔离计数，合法记录重新序列化后下推——下游拿到的都是经过类型核验的干净数据，内存占用恒定。

技术要点

• 缓冲重组保证跨块行边界正确
• schema 校验失败与 JSON 解析失败都隔离
• 对象模式与字节模式分段衔接

代码解读

ValidatingLines 的 _transform 把块并入缓冲再按换行切出完整行；每行先 JSON.parse 再过 RecordSchema 的三个约束，通过则计数并序列化下推，任一步失败计入 quarantined。尾部无换行的残行在流结束后按需处理。统计字段暴露在实例上供管道收尾报告。

使用建议

隔离行应连同原始内容写进死信文件供人工修复。大文件配合断点续传记录已处理偏移。schema 演进时隔离率突变是数据异常的早期信号。

代码示例

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

const RecordSchema = z.object({
  userId: z.number().int().positive(),
  event: z.string().min(1),
  at: z.string().datetime(),
});

class ValidatingLines extends Transform {
  private buffer = '';
  public valid = 0;
  public quarantined = 0;

  constructor() {
    super({ objectMode: true, readableObjectMode: false });
  }

  _transform(chunk: Buffer, _enc: string, callback: (err?: Error | null, data?: never) => void) {
    this.buffer += chunk.toString('utf8');
    let newline: number;

    while ((newline = this.buffer.indexOf('\n')) !== -1) {
      const line = this.buffer.slice(0, newline).trim();
      this.buffer = this.buffer.slice(newline + 1);
      if (!line) continue;

      try {
        const parsed = RecordSchema.parse(JSON.parse(line));
        this.valid += 1;
        this.push(JSON.stringify(parsed) + '\n');
      } catch {
        this.quarantined += 1;
      }
    }
    callback();
  }
}

const validator = new ValidatingLines();

await pipeline(
  createReadStream('events.ndjson'),
  validator,
  async function* (records) {
    for await (const line of records) {
      if (validator.valid % 100 === 0) console.log('streamed', validator.valid);
    }
  },
);

console.log('valid:', validator.valid, 'quarantined:', validator.quarantined);