Streaming baris per baris, batching, backpressure dan pipeline

Sebelumnya, di lesson 0014: jangan pernah memegang seluruh file; parse satu resource setiap kali. Lesson ini menunjukkan caranya: baca per baris, kumpulkan per batch, dan biarkan backpressure menahan pembacaan saat database lebih lambat. Seperti seluruh modul ini, tanpa lab: potongan kode hanya untuk dibaca dan dijelaskan.

1. Streaming baris per baris, batching, dan stream-json

Versi paling pendek, persis pola resmi dokumentasi readline "Read file stream line-by-line": ^[1]

import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";

const rl = createInterface({ input: createReadStream("Observation_1.ndjson"), crlfDelay: Infinity });
let batch: Observation[] = [];
for await (const line of rl) {
  if (!line.trim()) continue;
  batch.push(JSON.parse(line));                     // satu resource kecil, bukan seluruh file
  if (batch.length === 1000) { await db.insertMany(batch); batch = []; }
}
if (batch.length) await db.insertMany(batch);

• crlfDelay: Infinity membuat CR LF (\r\n) dihitung satu pemisah baris; relevan karena nd-json FHIR menyebut "newline pair (characters 13 and 10)". ^{[1] [2]}
• Dua catatan dari dokumentasi async iterator readline: "Errors in the input stream are not forwarded", dan "Performance is not on par with the traditional 'line' event API". ^[1] Dokumentasi juga tidak menjanjikan input dijeda selama Anda await di dalam loop. Untuk job produksi, rangkai dengan pipeline (bagian 2).
• Batching (mis. 1.000 resource per insert) mengurangi round-trip ke database; await pada insert adalah tempat "konsumen lambat" yang harus menahan pembacaan file.

JSON besar yang bukan NDJSON (mis. satu Bundle searchset 2 GB dari sistem lain) tidak bisa dibaca per baris, karena JSON biasa boleh punya newline di mana saja. Untuk itu ada stream-json: "Point it at a document far larger than available memory and it streams straight through", dan komponennya dirangkai dengan stream-chain yang "handles the streaming and backpressure". README-nya juga mengingatkan: dirancang untuk "data you own or trust", bukan input publik yang tidak tepercaya. ^[3]

import { parser } from "stream-json";
import { pick } from "stream-json/filters/pick.js";
import { streamArray } from "stream-json/streamers/stream-array.js";
import chain from "stream-chain";
import fs from "node:fs";

const entries = chain([
  fs.createReadStream("searchset-besar.json"),     // { "resourceType": "Bundle", "entry": [ ...jutaan... ] }
  parser(),
  pick({ filter: "entry" }),                        // turun ke array "entry", abaikan sisanya
  streamArray(),                                    // satu elemen array setiap kali: { key, value }
]);
entries.on("data", ({ value }) => handle(value.resource));

Pola dan import mengikuti contoh README stream-json 3.x (ESM, pick lalu streamArray). ^[3] README yang sama menyebut parser JSONL di stream-chain sebagai "the fast path for line-delimited data dumps".

Membaca per baris sudah benar, tapi apa yang mencegah memori menumpuk saat database lebih lambat dari disk? Itu tugas backpressure.

2. Backpressure dan pipeline

Disk bisa memberi 500 MB/detik, database mungkin hanya menerima beberapa ribu baris per detik. Tanpa rem, data menumpuk di memori. Panduan Node menyebut ini backpressure: "a buildup of data behind a buffer during data transfer", dan di Node "streams have been the adopted solution". ^[4]

• highWaterMark: setelah buffer baca mencapai ambang ini, stream "will temporarily stop reading data from the underlying resource until the data currently buffered can be consumed". Dalam object mode, ambangnya dihitung dalam jumlah objek. Ini "a threshold, not a limit". ^[5]
• write() → false: "Once write() returns false, do not write more chunks until the 'drain' event is emitted." Kalau diabaikan, "Node.js will buffer all written chunks until maximum memory usage occurs, at which point it will abort unconditionally." ^[5]
• pipeline vs .pipe(): pipeline "pipe[s] between streams and generators forwarding errors and properly cleaning up". Dengan .pipe(), "if the Readable stream emits an error during processing, the Writable destination is not closed automatically". ^{[5] [5]}

Versi produksi: unduh dari URL manifest, pecah per baris, batch ke database, semua dalam satu pipeline yang bisa di-await. pipeline dari node:stream/promises menerima async generator sebagai tahap transform dan fungsi async sebagai tujuan. ^[5]

import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";

async function* lines(source: AsyncIterable<string>) {
  let rest = "";
  for await (const chunk of source) {               // pull: chunk berikut dibaca setelah body loop selesai
    const parts = (rest + chunk).split("\n");
    rest = parts.pop()!;                            // sisa baris yang belum lengkap
    yield* parts;
  }
  if (rest) yield rest;
}

async function importFile(url: string, token: string) {
  const res = await fetch(url, { headers: { Accept: "application/fhir+ndjson", Authorization: `Bearer ${token}` } });
  if (!res.ok || !res.body) throw new Error(`unduh gagal: ${res.status}`);
  await pipeline(
    Readable.fromWeb(res.body).setEncoding("utf8"), // web stream → Node stream, decode UTF-8 dengan aman
    lines,
    async function (source: AsyncIterable<string>) {
      let batch: unknown[] = [];
      for await (const line of source) {
        if (!line.trim()) continue;
        batch.push(JSON.parse(line));
        if (batch.length === 1000) { await db.insertMany(batch); batch = []; }   // insert lambat = baca berhenti
      }
      if (batch.length) await db.insertMany(batch);
    },
  );                                                // error di tahap mana pun → reject + semua stream dibersihkan
}

Kenapa memori tetap kecil: async iterator stream bersifat pull. Selama await db.insertMany belum selesai, tidak ada yang meminta baris berikutnya, buffer terisi sampai highWaterMark, lalu pembacaan dari jaringan berhenti. Iterasi "will be read in chunks of size equal to the highWaterMark option". ^[5] setEncoding("utf8") mencegah karakter multibyte (mis. nama pasien beraksen) terpotong di batas chunk.

Berikutnya, lesson 0016: satu jebakan khas FHIR saat mem-parse baris-baris itu, yaitu angka decimal yang kehilangan presisi di JavaScript.

Istilah: GLOSSARY.md (bagian "Bulk Data dan JSON skala besar") · Indeks: semua lesson FHIR

Ada yang belum jelas? Tanyakan ke saya (agent) di terminal, misalnya: "apa beda Patient/$export dan Group/[id]/$export untuk use case analitik?", "bagaimana menulis hasil export ke S3 dengan backpressure yang benar?", atau "kapan lebih baik memakai worker thread daripada streaming?"

Sumber: [1] Node.js: Readline · [2] FHIR R4: ND-JSON Representation of Resources · [3] stream-json README · [4] Node.js Learn: Backpressuring in Streams · [5] Node.js: Stream