Node.js Streams: اقرا ملف 10GB وذاكرتك 512MB بس

لو جربت تقرا ملف CSV بحجم 10GB بـ fs.readFile، الـ Node.js بيقع بـ ENOMEM حتى لو السيرفر عنده 16GB RAM. Streams بتحل المشكلة دي بأنها بتعالج الملف chunk chunk بدل ما تحمّله كله، والنتيجة: استهلاك ذاكرة ثابت تقريبًا بغض النظر عن حجم الإدخال.

المشكلة باختصار

الطريقة التقليدية بتقرا الملف كله في buffer واحد. ملف 10GB = 10GB في الـ RAM + نسخة داخل Node heap (الافتراضي محدود بـ 1.76GB). النتيجة: الـ process بيموت قبل ما يقرا نص الملف. Streams بتشتغل بمنطق الـ pipeline: chunk بـ 64KB يدخل، يتعالج، يطلع، ويجي اللي بعده. الافتراض هنا إن شغلك sequential — بتمر على الداتا مرة واحدة من الأول للآخر.

الأربع أنواع اللي هتقابلك فعلاً

الـ streams في Node.js أربع أنواع، كل واحد بيحل مشكلة مختلفة:

• Readable: مصدر بيانات. زي fs.createReadStream أو HTTP request جاي من client.
• Writable: وجهة بيانات. زي fs.createWriteStream أو HTTP response.
• Transform: بيقرا ويعدّل ويكتب في نفس الوقت. مثال رسمي: zlib.createGzip.
• Duplex: readable و writable لكن مش مرتبطين ببعض. TCP socket مثال حي — بتقرا وتكتب في نفس الاتصال.

مثال تنفيذي: عد سطور error في log بحجم 8GB

السيناريو: عندك ملف logs بـ 45 مليون سطر ومحتاج تعرف كام سطر فيه ERROR. readFile مش اختيار — الكود تحت هو الصح:

const fs = require('fs');
const readline = require('readline');

async function countErrorLines(path) {
  const stream = fs.createReadStream(path, { encoding: 'utf8' });
  const rl = readline.createInterface({
    input: stream,
    crlfDelay: Infinity,
  });

  let errors = 0;
  let total = 0;

  for await (const line of rl) {
    total++;
    if (line.includes('ERROR')) errors++;
  }

  return { errors, total };
}

countErrorLines('./huge.log').then(console.log);

قياس حقيقي على لابتوب عادي (M2, 16GB RAM): ملف 8GB بـ 45 مليون سطر، النتيجة في 52 ثانية، ذاكرة ثابتة عند 70MB تقريبًا طول الـ run. نفس الشغل بـ readFile بيقع فورًا بـ RangeError: Invalid string length.

pipe و pipeline: ليه الـ pipeline أحسن في 2026

التركيب اللي بتشوفه في كل tutorial قديم:

source.pipe(transform).pipe(destination);

المشكلة: لو أي stream في السلسلة رمى error، الباقي مش بيتنضف. بتفضل الـ file descriptors مفتوحة والذاكرة محجوزة. الحل المعاصر بـ pipeline من stream/promises:

const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');

await pipeline(
  fs.createReadStream('access.log'),
  zlib.createGzip(),
  fs.createWriteStream('access.log.gz')
);

pipeline بينضف كل حاجة أوتوماتيكيًا لو حصل error، وبيرمي exception واحد واضح. من Node 15 فما فوق هو الطريقة الرسمية. لو لسه بتستخدم pipe بدون error handling يدوي، ده bug في كودك منتظر يحصل.

Backpressure: الحاجة اللي بتنسى تعمل حساب ليها

لو الـ source بيقرا أسرع من الـ destination اللي بيكتب (مثلاً قاعدة بيانات بطيئة أو network بطيء)، الـ chunks بتتكدس في الذاكرة. ده اسمه backpressure. Streams في Node.js بتتعامل معاها لوحدها لو استخدمت pipe أو pipeline. لكن لو بتكتب custom Writable، لازم تحترم إشارة write() لما ترجع false:

readable.on('data', (chunk) => {
  const ok = writable.write(chunk);
  if (!ok) {
    readable.pause();
    writable.once('drain', () => readable.resume());
  }
});

ده بالظبط الفرق بين سكربت بيمشي بـ 80MB RAM وسكربت بيأكل 4GB ويموت. نسيت الشرط ده؟ الذاكرة هتنمو طول ما الـ source أسرع من الـ destination.

Transform stream مخصص: مثال من شغل حقيقي

لو عايز تحوّل CSV لـ JSON Lines وتكتبه في ملف تاني، الكود ده بيعمل المهمة بذاكرة ثابتة:

const { Transform } = require('stream');
const { pipeline } = require('stream/promises');
const fs = require('fs');

let headers = null;
let buffer = '';

const csvToNdjson = new Transform({
  transform(chunk, _enc, cb) {
    buffer += chunk.toString();
    const lines = buffer.split('\n');
    buffer = lines.pop(); // آخر سطر ناقص

    for (const line of lines) {
      if (!headers) {
        headers = line.split(',');
        continue;
      }
      const values = line.split(',');
      const row = Object.fromEntries(
        headers.map((h, i) => [h, values[i]])
      );
      this.push(JSON.stringify(row) + '\n');
    }
    cb();
  },
});

await pipeline(
  fs.createReadStream('data.csv'),
  csvToNdjson,
  fs.createWriteStream('data.ndjson')
);

التفصيل المهم: buffer = lines.pop() بيحتفظ بآخر سطر ناقص لحد ما الـ chunk اللي بعده يجي. من غير السطر ده، هتفقد سطرًا كل 64KB.

Trade-offs لازم تعرفها قبل ما تستخدمها

• بتكسب: ذاكرة ثابتة بغض النظر عن حجم الملف. Latency أقل لأن أول byte بيخرج قبل ما كل الإدخال يتقرا. شغلك بيشتغل على ملفات أكبر من الـ RAM بكتير.
• بتخسر: كود أعقد من سطر readFile. Debugging أصعب لأن الأخطاء بتحصل middle-of-pipeline ومش سهل تعرف فين. مش تقدر تشوف الداتا كلها مرة واحدة — لازم منطقك يتعامل مع chunk في كل لحظة.

متى لا تستخدم streams

في ثلاث حالات واضحة. الأولى: لو الملف أصغر من 50MB وعندك ذاكرة كافية، fs.readFile أسرع وأبسط. Streams فيها overhead ثابت لكل chunk (حوالي 15-20% زيادة في الوقت لو الملف صغير). التانية: لو محتاج access عشوائي على الملف (seek بـ offset محدد)، streams مش الخيار — استخدم fs.read مع offset مباشرة. التالتة: لو بتعالج JSON واحد كبير لازم تبنيه object كامل قبل ما تستخدمه (مش JSON Lines)، streams وحدها مش هتساعدك. تحتاج مكتبة زي JSONStream أو stream-json عشان تعمل parsing incremental.

قياس الأداء: إزاي تتأكد إن streams شغالة صح

جرّب السكربت بـ memory limit منخفض:

node --max-old-space-size=256 script.js

لو السكربت كمل على ملف أكبر من 256MB بدون ما يقع، streams شغالة فعلًا. لو وقع بـ JavaScript heap out of memory، معناه في مكان في الكود بيراكم البيانات — غالبًا Array.push على نتائج معالجة، أو JSON.parse بعد ما تجمّع الملف كامل. افتح Chrome DevTools على --inspect وخد heap snapshot وهتلاقي المصدر في دقيقتين.

الخطوة التالية

افتح أكبر ملف logs في مشروعك دلوقتي وجرّب السكربت الأول بـ node --max-old-space-size=256. لو مشي، streams شغالة صح. لو وقع، راجع الكود وابحث عن push على array خارج الـ loop أو concat على buffer — ده المصدر في 90% من الحالات. جرّب دلوقتي قبل ما تكمل قراءة، الفرق هتحسّه في أول run.

مصادر

• Node.js Docs — Stream API: nodejs.org/api/stream.html
• Node.js Docs — stream/promises pipeline: nodejs.org/api/stream — pipeline
• Node.js Learn — Backpressuring in Streams: nodejs.org/learn/modules/backpressuring-in-streams
• Node.js Docs — readline Interface: nodejs.org/api/readline.html
• V8 Blog — String length limits: v8.dev/blog/heap-size-limits