[NODE JS] Readable Stream ile Büyük Veri Okuma

Silverhand · Pazartesi saat 10:44'de

[NODE JS] Readable Stream ile Büyük Veri Okuma

Node.js'in en güçlü özelliklerinden biri stream'lerdir. Büyük dosyaları veya veri kümelerini işlerken tüm veriyi belleğe yüklemek yerine, parça parça işlemek hem bellek kullanımını düşürür hem de ilk yanıt süresini kısaltır. Readable Stream, bu yaklaşımın okuma tarafını temsil eder ve Node.js'te her yerde karşınıza çıkar.

İlk kez stream'lerin gücünü anladığım an, 2 GB boyutunda bir CSV dosyasını işlemeye çalıştığımdı. fs.readFile ile dosyayı okumaya çalıştığımda Node.js bellek hatası verdi. Readable Stream kullanarak aynı dosyayı satır satır işlediğimde, bellek kullanımı 50 MB'ın altında kaldı. Bu deneyim, stream'lerin neden bu kadar önemli olduğunu bana somut olarak gösterdi.

Node.js'te Readable Stream'in iki çalışma modu vardır: flowing (akan) ve paused (duraklatılmış). Flowing modda veri otomatik olarak akar ve data olayı ile yakalanır. Paused modda ise read() metodu ile manuel olarak veri çekilir. Genellikle flowing mod daha yaygın kullanılır, ancak veri akış hızını kontrol etmeniz gereken durumlarda paused mod tercih edilebilir.

Backpressure (geri basınç) kavramı, stream'lerin en kritik konularından biridir. Veri kaynağı, tüketiciden daha hızlı veri ürettiğinde, bellekte birikim oluşur ve sonunda uygulama çökebilir. Node.js stream'leri bu durumu otomatik olarak yönetir. Tüketici yavaş kaldığında, Readable Stream veri üretimini geçici olarak durdurur. Bu mekanizma, uygulamanızın büyük veri kümeleriyle bile kararlı çalışmasını sağlar.

Kendi Readable Stream'inizi oluşturmak da oldukça kolaydır. stream.Readable sınıfını genişleterek ve _read metodunu implement ederek, herhangi bir veri kaynağından stream oluşturabilirsiniz. Veritabanı sorguları, API yanıtları, sensör verileri gibi farklı kaynaklardan gelen verileri stream haline getirmek, uygulamanızın modülerliğini artırır.

HTTP isteklerinin kendisi de birer Readable Stream'dir. Express'te req nesnesi bir Readable Stream olduğu için, büyük dosya yüklemelerini parça parça işleyebilirsiniz. Bu özellikle video veya büyük resim dosyalarını işlerken çok kullanışlıdır. Tüm dosyayı belleğe almadan, gelen parçaları direkt diske yazabilir veya başka bir servise aktarabilirsiniz.

Readable Stream'lerin objectMode özelliği de bilinmesi gereken önemli bir konfigürasyondur. Varsayılan olarak stream'ler Buffer veya string verilerle çalışır. objectMode aktif edildiğinde, JavaScript nesnelerini stream üzerinden aktarabilirsiniz. Bu özellik, veri dönüşüm pipeline'ları oluştururken son derece kullanışlıdır.

JavaScript:

// Readable Stream ile Büyük Veri Okuma
const { Readable } = require('stream');
const fs = require('fs');
const path = require('path');
const readline = require('readline');

// ==========================================
// 1. Dosyadan Readable Stream ile Okuma
// ==========================================
function dosyadanOku(dosyaYolu) {
  console.log('--- Dosyadan Stream Okuma ---');

  const okumaAkisi = fs.createReadStream(dosyaYolu, {
    encoding: 'utf-8',
    highWaterMark: 16 * 1024 // 16KB parçalar halinde oku
  });

  let toplamBoyut = 0;
  let parcaSayisi = 0;

  okumaAkisi.on('data', (parca) => {
    parcaSayisi++;
    toplamBoyut += parca.length;
    console.log('Parça ' + parcaSayisi + ': ' + parca.length + ' karakter okundu');
  });

  okumaAkisi.on('end', () => {
    console.log('Okuma tamamlandı!');
    console.log('Toplam boyut: ' + toplamBoyut + ' karakter');
    console.log('Toplam parça: ' + parcaSayisi);
  });

  okumaAkisi.on('error', (hata) => {
    console.error('Okuma hatası:', hata.message);
  });
}

// ==========================================
// 2. Satır Satır Okuma (CSV İşleme)
// ==========================================
async function satirSatirOku(dosyaYolu) {
  console.log('\n--- Satır Satır CSV Okuma ---');

  const akis = fs.createReadStream(dosyaYolu, { encoding: 'utf-8' });
  const satirOkuyucu = readline.createInterface({
    input: akis,
    crlfDelay: Infinity
  });

  let satirSayisi = 0;
  let basliklar = [];
  const veriler = [];

  for await (const satir of satirOkuyucu) {
    satirSayisi++;

    if (satirSayisi === 1) {
      // İlk satır başlıklar
      basliklar = satir.split(',').map(b => b.trim());
      console.log('Başlıklar:', basliklar);
      continue;
    }

    // Veriyi objeye dönüştür
    const degerler = satir.split(',').map(d => d.trim());
    const kayit = {};
    basliklar.forEach((baslik, indeks) => {
      kayit[baslik] = degerler[indeks] || '';
    });

    veriler.push(kayit);

    // Her 1000 satırda ilerleme bildirimi
    if (satirSayisi % 1000 === 0) {
      console.log(satirSayisi + ' satır işlendi...');
    }
  }

  console.log('Toplam ' + satirSayisi + ' satır okundu');
  console.log('İlk kayıt:', veriler[0]);
  return veriler;
}

// ==========================================
// 3. Özel Readable Stream Oluşturma
// ==========================================
class VeritabaniAkisi extends Readable {
  constructor(sorgu, secenekler = {}) {
    super({
      objectMode: true, // JavaScript nesneleri aktarabilmek için
      highWaterMark: secenekler.parcaBoyutu || 100
    });
    this.sorgu = sorgu;
    this.sayfa = 0;
    this.sayfaBoyutu = secenekler.sayfaBoyutu || 100;
    this.bpitiMi = false;
  }

  // _read metodu stream tarafından otomatik çağrılır
  async _read() {
    if (this.bitiMi) {
      this.push(null); // Stream'i sonlandır
      return;
    }

    try {
      // Veritabanından sayfalı veri çekme (simülasyon)
      const veriler = await this.veritabanindanGetir(this.sayfa, this.sayfaBoyutu);

      if (veriler.length === 0) {
        this.bitiMi = true;
        this.push(null);
        return;
      }

      // Her kaydı stream'e gönder
      for (const kayit of veriler) {
        // push false dönerse, tüketici hazır değil demektir
        const devamEt = this.push(kayit);
        if (!devamEt) {
          break; // Backpressure - bekle
        }
      }

      this.sayfa++;
    } catch (hata) {
      this.destroy(hata);
    }
  }

  // Sahte veritabanı sorgusu
  async veritabanindanGetir(sayfa, boyut) {
    // Gerçek projede burada veritabanı sorgusu olur
    const toplamKayit = 500;
    const baslangic = sayfa * boyut;

    if (baslangic >= toplamKayit) {
      return [];
    }

    const sonuc = [];
    const bitis = Math.min(baslangic + boyut, toplamKayit);

    for (let i = baslangic; i < bitis; i++) {
      sonuc.push({
        id: i + 1,
        isim: 'Ürün ' + (i + 1),
        fiyat: Math.floor(Math.random() * 10000) / 100,
        kategori: ['Elektronik', 'Giyim', 'Kitap', 'Spor'][i % 4],
        olusturulma: new Date(Date.now() - i * 86400000).toISOString()
      });
    }

    // Veritabanı gecikmesini simüle et
    await new Promise(resolve => setTimeout(resolve, 50));
    return sonuc;
  }
}

// ==========================================
// 4. Readable.from ile Kolay Stream Oluşturma
// ==========================================
function iteratordenStream() {
  console.log('\n--- Iterator'den Stream ---');

  // Generator fonksiyonu ile veri üretme
  function* veriUretici() {
    const sehirler = [
      { isim: 'İstanbul', nufus: 16000000 },
      { isim: 'Ankara', nufus: 5700000 },
      { isim: 'İzmir', nufus: 4400000 },
      { isim: 'Bursa', nufus: 3100000 },
      { isim: 'Antalya', nufus: 2600000 },
      { isim: 'Adana', nufus: 2200000 },
      { isim: 'Konya', nufus: 2300000 },
      { isim: 'Gaziantep', nufus: 2100000 }
    ];

    for (const sehir of sehirler) {
      yield JSON.stringify(sehir) + '
';
    }
  }

  const akis = Readable.from(veriUretici());

  akis.on('data', (parca) => {
    const veri = JSON.parse(parca.toString().trim());
    console.log(veri.isim + ': ' + veri.nufus.toLocaleString('tr-TR') + ' kişi');
  });

  akis.on('end', () => {
    console.log('Şehir verileri tamamlandı');
  });
}

// ==========================================
// 5. Async Iterator ile Stream Tüketme
// ==========================================
async function asyncIteratorKullanimi() {
  console.log('\n--- Async Iterator ile Tüketme ---');

  const dbAkisi = new VeritabaniAkisi('SELECT * FROM urunler', {
    sayfaBoyutu: 50
  });

  let sayac = 0;
  let toplamFiyat = 0;
  const kategoriler = {};

  // for await...of ile stream tüketme (modern yaklaşım)
  for await (const kayit of dbAkisi) {
    sayac++;
    toplamFiyat += kayit.fiyat;

    // Kategori sayımı
    kategoriler[kayit.kategori] = (kategoriler[kayit.kategori] || 0) + 1;
  }

  console.log('İşlenen kayıt sayısı:', sayac);
  console.log('Ortalama fiyat:', (toplamFiyat / sayac).toFixed(2) + ' TL');
  console.log('Kategori dağılımı:', kategoriler);
}

// ==========================================
// Ana Çalıştırıcı
// ==========================================
async function basla() {
  try {
    // Iterator'den stream
    iteratordenStream();

    // Async iterator kullanımı
    await asyncIteratorKullanimi();

    console.log('\nTüm stream işlemleri tamamlandı');
  } catch (hata) {
    console.error('Hata:', hata.message);
  }
}

basla();

Readable Stream'ler, büyük veri kümeleriyle çalışırken vazgeçilmez bir araçtır. Bellek verimli, performanslı ve ölçeklenebilir uygulamalar geliştirmek istiyorsanız, stream kavramını derinlemesine öğrenmenizi şiddetle tavsiye ederim. Bir sonraki yazıda Writable Stream'leri inceleyeceğiz.

Best Studio · Çarşamba saat 16:17'de

Emeğiniz ve katkınız için teşekkür ederiz!
Konunuz gerekli incelemelerden geçerek onaylandı. Topluluğumuza katkılarınızın devamını bekliyoruz.
İyi forumlar!

[NODE JS] Readable Stream ile Büyük Veri Okuma

Silverhand