[NODE JS] Transform Stream ile Veri Dönüştürme

Silverhand · Pazartesi saat 10:44'de

[NODE JS] Transform Stream ile Veri Dönüştürme

Transform Stream, hem okunabilir hem de yazılabilir olan özel bir stream türüdür. Gelen veriyi alır, üzerinde bir dönüşüm uygular ve çıkışa gönderir. Bir nevi veri işleme hattındaki ara istasyon gibi düşünebilirsiniz. CSV'yi JSON'a çevirmek, metni şifrelemek, veriyi sıkıştırmak veya log formatını değiştirmek gibi işlemlerin hepsi Transform Stream ile zarif şekilde çözülebilir.

Transform Stream'in en büyük avantajı, veri işleme mantığını modüler hale getirmesidir. Her dönüşüm adımını ayrı bir Transform Stream olarak yazabilir ve bunları pipe ile birbirine bağlayabilirsiniz. Bu yaklaşım, Unix felsefesindeki "tek bir işi iyi yap" ilkesini Node.js dünyasına taşır. Küçük, odaklı dönüşüm modülleri yazarak karmaşık veri işleme pipeline'ları oluşturabilirsiniz.

İlk Transform Stream'imi bir log analiz projesi için yazdım. Ham log dosyalarını okuyup, her satırı ayrıştırıyor, IP adreslerini gizliyor, tarih formatını değiştiriyor ve sonucu yeni bir dosyaya yazıyordum. Her adımı ayrı bir Transform Stream olarak modellediğimde, kodun okunabilirliği ve test edilebilirliği inanılmaz arttı. İleride bir adımı değiştirmem gerektiğinde, sadece ilgili Transform'u güncellemem yeterliydi.

_transform metodu, Transform Stream'in kalbidir. Bu metot her veri parçası için çağrılır. Dönüştürülmüş veriyi this.push() ile çıkışa gönderirsiniz. Bir girişten birden fazla çıkış üretebilir veya bazı girdileri filtreleyerek hiç çıkış göndermeyebilirsiniz. Bu esneklik, Transform Stream'i son derece güçlü kılar.

_flush metodu ise stream sonlandırılmadan hemen önce çağrılır. Tampondaki kalan verileri işlemek için kullanılır. Örneğin, satır satır işleme yapıyorsanız ve son satırda yeni satır karakteri yoksa, _flush içinde bu son parçayı işleyebilirsiniz. Bu metodun doğru kullanımı, veri kaybını önlemek açısından kritiktir.

objectMode kullanarak Transform Stream'lerinizi nesne bazlı çalıştırabilirsiniz. Bu özellikle veri pipeline'larında çok kullanışlıdır. Örneğin, bir Transform bir JSON satırı alıp bir JavaScript nesnesine dönüştürür, sonraki Transform bu nesneyi zenginleştirir, bir sonraki filtreleme yapar. Her adım temiz ve bağımsızdır.

Performans açısından dikkat edilmesi gereken noktalar var. Transform Stream içinde ağır hesaplama veya asenkron işlemler yapıyorsanız, throughput düşebilir. Bu durumda parallel transform veya worker thread kullanımı düşünülebilir. Ancak çoğu senaryo için standart Transform Stream yeterli performansı sağlar.

JavaScript:

// Transform Stream ile Veri Dönüştürme
const { Transform, Readable, Writable, pipeline } = require('stream');
const { promisify } = require('util');
const zlib = require('zlib');
const crypto = require('crypto');

const pipelineAsync = promisify(pipeline);

// ==========================================
// 1. CSV'den JSON'a Dönüştürücü
// ==========================================
class CsvdenJsonaDonusturucu extends Transform {
  constructor(secenekler = {}) {
    super({
      readableObjectMode: true, // Çıkış: nesne
      writableObjectMode: false  // Giriş: string/buffer
    });
    this.ayirici = secenekler.ayirici || ',';
    this.basliklar = null;
    this.kalanVeri = '';
    this.satirSayisi = 0;
  }

  _transform(parca, kodlama, geriCagir) {
    // Önceki kalan veri ile birleştir
    const veri = this.kalanVeri + parca.toString();
    const satirlar = veri.split('
');

    // Son satır eksik olabilir, onu sakla
    this.kalanVeri = satirlar.pop() || '';

    for (const satir of satirlar) {
      const temizSatir = satir.trim();
      if (!temizSatir) continue;

      if (!this.basliklar) {
        // İlk satır başlıkları içerir
        this.basliklar = temizSatir.split(this.ayirici).map(b => b.trim());
        continue;
      }

      this.satirSayisi++;
      const degerler = temizSatir.split(this.ayirici).map(d => d.trim());
      const nesne = {};

      this.basliklar.forEach((baslik, indeks) => {
        let deger = degerler[indeks] || '';
        // Sayısal değerleri otomatik dönüştür
        if (!isNaN(deger) && deger !== '') {
          deger = Number(deger);
        }
        nesne[baslik] = deger;
      });

      this.push(nesne);
    }

    geriCagir();
  }

  _flush(geriCagir) {
    // Kalan veriyi işle
    if (this.kalanVeri.trim() && this.basliklar) {
      const degerler = this.kalanVeri.trim().split(this.ayirici).map(d => d.trim());
      const nesne = {};
      this.basliklar.forEach((baslik, indeks) => {
        nesne[baslik] = degerler[indeks] || '';
      });
      this.push(nesne);
    }
    console.log('CSV dönüşümü tamamlandı: ' + this.satirSayisi + ' satır işlendi');
    geriCagir();
  }
}

// ==========================================
// 2. Veri Filtreleme Transform'u
// ==========================================
class VeriFiltreleme extends Transform {
  constructor(filtreFonksiyonu) {
    super({ objectMode: true });
    this.filtre = filtreFonksiyonu;
    this.gecenSayisi = 0;
    this.elenenSayisi = 0;
  }

  _transform(kayit, kodlama, geriCagir) {
    if (this.filtre(kayit)) {
      this.gecenSayisi++;
      this.push(kayit);
    } else {
      this.elenenSayisi++;
    }
    geriCagir();
  }

  _flush(geriCagir) {
    console.log('Filtreleme sonucu - Geçen: ' + this.gecenSayisi
      + ', Elenen: ' + this.elenenSayisi);
    geriCagir();
  }
}

// ==========================================
// 3. Veri Zenginleştirme Transform'u
// ==========================================
class VeriZenginlestirme extends Transform {
  constructor(zenginlestirmeFn) {
    super({ objectMode: true });
    this.zenginlestir = zenginlestirmeFn;
  }

  async _transform(kayit, kodlama, geriCagir) {
    try {
      const zenginKayit = await this.zenginlestir(kayit);
      this.push(zenginKayit);
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }
}

// ==========================================
// 4. JSON Satır Dönüştürücü (NDJSON)
// ==========================================
class JsonSatirDonusturucu extends Transform {
  constructor() {
    super({
      writableObjectMode: true, // Giriş: nesne
      readableObjectMode: false  // Çıkış: string
    });
  }

  _transform(nesne, kodlama, geriCagir) {
    try {
      const jsonSatir = JSON.stringify(nesne) + '
';
      this.push(jsonSatir);
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }
}

// ==========================================
// 5. Şifreleme Transform'u
// ==========================================
class SifrelemeDonusturucu extends Transform {
  constructor(sifre, algoritma = 'aes-256-cbc') {
    super();
    this.algoritma = algoritma;
    const anahtar = crypto.scryptSync(sifre, 'tuz_degeri', 32);
    this.iv = crypto.randomBytes(16);
    this.sifreleyici = crypto.createCipheriv(algoritma, anahtar, this.iv);
    this.ilkParca = true;
  }

  _transform(parca, kodlama, geriCagir) {
    try {
      // İlk parçada IV'yi gönder
      if (this.ilkParca) {
        this.push(this.iv);
        this.ilkParca = false;
      }
      const sifrelenmis = this.sifreleyici.update(parca);
      this.push(sifrelenmis);
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }

  _flush(geriCagir) {
    try {
      const sonParca = this.sifreleyici.final();
      this.push(sonParca);
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }
}

// ==========================================
// 6. İstatistik Toplama Transform'u
// ==========================================
class IstatistikToplayici extends Transform {
  constructor(sayisalAlanlar = []) {
    super({ objectMode: true });
    this.sayisalAlanlar = sayisalAlanlar;
    this.istatistikler = {};
    this.toplamKayit = 0;

    // Her sayısal alan için istatistik başlat
    sayisalAlanlar.forEach(alan => {
      this.istatistikler[alan] = {
        toplam: 0,
        minimum: Infinity,
        maksimum: -Infinity,
        sayac: 0
      };
    });
  }

  _transform(kayit, kodlama, geriCagir) {
    this.toplamKayit++;

    // Sayısal alanların istatistiklerini güncelle
    this.sayisalAlanlar.forEach(alan => {
      const deger = parseFloat(kayit[alan]);
      if (!isNaN(deger)) {
        const ist = this.istatistikler[alan];
        ist.toplam += deger;
        ist.sayac++;
        ist.minimum = Math.min(ist.minimum, deger);
        ist.maksimum = Math.max(ist.maksimum, deger);
      }
    });

    // Veriyi değiştirmeden aktar
    this.push(kayit);
    geriCagir();
  }

  _flush(geriCagir) {
    // Ortalamaları hesapla
    this.sayisalAlanlar.forEach(alan => {
      const ist = this.istatistikler[alan];
      ist.ortalama = ist.sayac > 0
        ? (ist.toplam / ist.sayac).toFixed(2)
        : 0;
    });

    console.log('\n=== İstatistik Raporu ===');
    console.log('Toplam kayıt:', this.toplamKayit);
    Object.entries(this.istatistikler).forEach(([alan, ist]) => {
      console.log(alan + ':');
      console.log('  Ortalama: ' + ist.ortalama);
      console.log('  Min: ' + ist.minimum + ' / Max: ' + ist.maksimum);
      console.log('  Toplam: ' + ist.toplam.toFixed(2));
    });

    geriCagir();
  }
}

// ==========================================
// Kullanım Örneği - Pipeline
// ==========================================
async function basla() {
  // Sahte CSV verisi oluştur
  const csvVerisi = [
    'isim,yas,sehir,maas',
    'Ahmet Yılmaz,28,İstanbul,15000',
    'Ayşe Demir,34,Ankara,22000',
    'Can Kaya,25,İzmir,12000',
    'Deniz Çelik,41,Bursa,28000',
    'Elif Şahin,30,Antalya,18000',
    'Fatma Öztürk,27,Konya,14000',
    'Gökhan Arslan,38,İstanbul,35000',
    'Hülya Aydın,29,Ankara,19000',
    'İbrahim Güneş,45,İzmir,42000',
    'Jale Korkmaz,32,İstanbul,25000'
  ].join('
');

  // CSV stream kaynağı
  const csvKaynak = Readable.from([csvVerisi]);

  // Sonuçları toplayan Writable
  const sonuclar = [];
  const toplayici = new Writable({
    objectMode: true,
    write(kayit, kodlama, geriCagir) {
      sonuclar.push(kayit);
      geriCagir();
    }
  });

  // Pipeline: CSV -> JSON -> Filtrele -> Zenginleştir -> İstatistik -> Topla
  await pipelineAsync(
    csvKaynak,
    new CsvdenJsonaDonusturucu(),
    new VeriFiltreleme(kayit => kayit.maas > 15000),
    new VeriZenginlestirme(async (kayit) => {
      return {
        ...kayit,
        vergiDilimi: kayit.maas > 30000 ? 'Yüksek' : 'Normal',
        netMaas: (kayit.maas * 0.85).toFixed(2)
      };
    }),
    new IstatistikToplayici(['maas', 'yas']),
    toplayici
  );

  console.log('\nSonuçlar:');
  sonuclar.forEach(k => {
    console.log(k.isim + ' - ' + k.sehir + ' - Brüt: '
      + k.maas + ' TL - Net: ' + k.netMaas + ' TL');
  });
}

basla().catch(console.error);

Transform Stream, veri işleme pipeline'larının temel yapı taşıdır. Modüler, test edilebilir ve yeniden kullanılabilir dönüşüm bileşenleri oluşturarak, karmaşık veri işleme senaryolarını zarif şekilde çözebilirsiniz. Bir sonraki konuda pipe ile stream yönlendirmeyi derinlemesine 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] Transform Stream ile Veri Dönüştürme

Silverhand