پردازش JSONL در Node.js؛ کنترل فشار و لغو عملیات

نام فایل: jsonl-transform.mjs؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import { TextDecoder } from 'node:util';

export async function* parseJsonLines(source, { maxLineBytes = 4096, signal } = {}) {
  if (!Number.isInteger(maxLineBytes) || maxLineBytes < 1) throw new RangeError('Invalid limit');
  const decoder = new TextDecoder('utf-8', { fatal: true });
  let pending = Buffer.alloc(0);
  function parse(line) {
    if (line.length > maxLineBytes) throw new RangeError('Line exceeds byte limit');
    const text = decoder.decode(line).trim();
    return text ? JSON.parse(text) : undefined;
  }
  for await (const chunk of source) {
    signal?.throwIfAborted();
    pending = Buffer.concat([pending, Buffer.from(chunk)]);
    let end;
    while ((end = pending.indexOf(10)) !== -1) {
      const value = parse(pending.subarray(0, end));
      pending = pending.subarray(end + 1);
      if (value !== undefined) yield value;
    }
    if (pending.length > maxLineBytes) throw new RangeError('Line exceeds byte limit');
  }
  signal?.throwIfAborted();
  if (pending.length) {
    const value = parse(pending);
    if (value !== undefined) yield value;
  }
}

این ماژول کامل، خط‌ها را برحسب بایت محدود و UTF-8 را strict بررسی می‌کند. highWaterMark سقف کل حافظه نیست؛ منبع این مثال chunkهای ۱۶ کیلوبایتی دارد.

نام فایل: export-jsonl.mjs؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import { createReadStream, createWriteStream } from 'node:fs';
import { link, unlink } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import { pipeline } from 'node:stream/promises';
import { parseJsonLines } from './jsonl-transform.mjs';

export async function exportJsonl(input, output, { signal, maxLineBytes = 4096 } = {}) {
  const temporary = `${output}.pending-${randomUUID()}`;
  try {
    await pipeline(
      createReadStream(input, { highWaterMark: 16 * 1024 }),
      async function* (source, { signal: pipelineSignal }) {
        for await (const row of parseJsonLines(source, { maxLineBytes, signal: pipelineSignal })) {
          if (!row || !Number.isInteger(row.id) || typeof row.name !== 'string') {
            throw new TypeError('Each row needs an integer id and a string name');
          }
          yield JSON.stringify({ id: row.id, name: row.name.trim() }) + '\n';
        }
      },
      createWriteStream(temporary, { flags: 'wx' }),
      { signal },
    );
    // Same-directory hard link publishes a completed file without overwriting.
    await link(temporary, output);
  } finally {
    await unlink(temporary).catch(error => {
      if (error.code !== 'ENOENT') throw error;
    });
  }
}

خروجی موجود بازنویسی نمی‌شود. hard link در همان پوشه به فایل کامل اشاره می‌کند و فایل موقت در پایان پاک می‌شود؛ برای فایل‌سیستم فاقد این قابلیت سیاست انتشار دیگری لازم است.

نام فایل: test-jsonl.mjs؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import test from 'node:test';
import assert from 'node:assert/strict';
import { Readable } from 'node:stream';
import { mkdtemp, writeFile, readFile, readdir, unlink, rmdir } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { parseJsonLines } from './jsonl-transform.mjs';
import { exportJsonl } from './export-jsonl.mjs';

async function collect(chunks, options) {
  const rows = [];
  for await (const row of parseJsonLines(Readable.from(chunks), options)) rows.push(row);
  return rows;
}
test('UTF-8 characters may span two chunks', async () => {
  const bytes = Buffer.from('{"id":1,"name":"لیان"}\n');
  assert.deepEqual(await collect([bytes.subarray(0, 18), bytes.subarray(18)]),
                   [{ id: 1, name: 'لیان' }]);
});
test('oversized lines, malformed JSON and invalid UTF-8 fail', async () => {
  await assert.rejects(collect([Buffer.from('x'.repeat(20))], { maxLineBytes: 8 }), RangeError);
  await assert.rejects(collect([Buffer.from('{bad}\n')]), SyntaxError);
  await assert.rejects(collect([Buffer.from([0xff, 10])]), TypeError);
});
test('completed output is exclusive; cancellation leaves no temporary output', async () => {
  const dir = await mkdtemp(join(tmpdir(), 'liyan-jsonl-test-'));
  try {
    const input = join(dir, 'input.jsonl'), output = join(dir, 'output.jsonl');
    await writeFile(input, '{"id":7,"name":" Liyan "}\n');
    await exportJsonl(input, output);
    const result = '{"id":7,"name":"Liyan"}\n';
    assert.equal(await readFile(output, 'utf8'), result);
    await assert.rejects(exportJsonl(input, output), { code: 'EEXIST' });
    assert.equal(await readFile(output, 'utf8'), result);
    const controller = new AbortController(); controller.abort();
    await assert.rejects(exportJsonl(input, join(dir, 'cancelled.jsonl'), { signal: controller.signal }));
    assert.deepEqual((await readdir(dir)).sort(), ['input.jsonl', 'output.jsonl']);
  } finally {
    for (const name of await readdir(dir)) await unlink(join(dir, name));
    await rmdir(dir); // Non-recursive: unexpected subdirectories are never removed.
  }
});

node --test test-jsonl.mjs را اجرا کنید. آزمون‌ها مرز نویسه، اندازهٔ خط، دادهٔ خراب، خروجی موجود و پاک‌ماندن نام‌های موقت پس از لغو را بررسی می‌کنند.

نام فایل: demo-jsonl.mjs؛ این فایل را کامل و بدون حذف سطرها ذخیره کنید.

import { mkdtemp, writeFile, readFile } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { exportJsonl } from './export-jsonl.mjs';

const dir = await mkdtemp(join(tmpdir(), 'liyan-jsonl-demo-'));
const input = join(dir, 'input.jsonl'), output = join(dir, 'output.jsonl');
await writeFile(input, '{"id":1,"name":" Liyan "}\n{"id":2,"name":" Data "}\n', { flag: 'wx' });
await exportJsonl(input, output);
console.log(await readFile(output, 'utf8'));
console.log('Inspect the input and output in:', dir);

با node demo-jsonl.mjs دو رکورد با name بدون فاصلهٔ اضافی چاپ می‌شود. پوشهٔ موقت برای بررسی فایل‌های ورودی و خروجی در پیام آخر معرفی می‌شود.