[NODE JS] Büyük JSON Dosyalarını Stream ile Parse Etme

[Silverhand]

Türkiye'de ilk Mobil & PC Aynı Anda Metin2 Oyna. Triarchonline kalıcı ve uzun ömürlü yapısı ile 24 Temmuz'da açılıyor. | 1-99 Mobil Metin2 Triarch HEMEN TIKLA!

[NODE JS] Büyük JSON Dosyalarını Stream ile Parse Etme​

JSON, modern web geliştirmenin en yaygın veri formatıdır. API yanıtları, yapılandırma dosyaları, veri yedekleri ve daha pek çok alanda JSON kullanılır. Küçük JSON dosyaları için JSON.parse() gayet yeterlidir; ancak dosya boyutu megabaytları hatta gigabaytları bulduğunda işler karmaşıklaşır. Ben bir kez bir analitik projesinde 2 GB büyüklüğünde bir JSON dosyasını parse etmem gerektiğinde, standart yöntemlerin ne kadar yetersiz kaldığını acı bir şekilde deneyimledim.

JSON.parse() fonksiyonu tüm veriyi tek seferde belleğe yükler ve işler. 2 GB'lık bir JSON dosyası için Node.js'in varsayılan bellek limiti olan 1.5 GB bile yetmez. --max-old-space-size parametresiyle bellek limitini artırabilirsiniz ama bu gerçek bir çözüm değildir. Çünkü parse işlemi sırasında orijinal string ve oluşturulan JavaScript nesneleri aynı anda bellekte yer kaplar, bu da dosya boyutunun iki-üç katı kadar bellek tüketimi anlamına gelir.

Stream tabanlı JSON parse yaklaşımı bu sorunu köklü bir şekilde çözer. Dosyayı parça parça okuyarak JSON yapısındaki her öğeyi ayrı ayrı işlersiniz. Büyük bir dizi içindeki her nesneyi bağımsız olarak parse eder ve işlersiniz. Bu sayede bellekte aynı anda sadece bir veya birkaç nesne bulunur.

Node.js ekosisteminde bu iş için JSONStream, stream-json ve clarinet gibi popüler kütüphaneler mevcuttur. Ancak temel mantığı anlamak için kendi basit stream parser'ınızı da yazabilirsiniz. Transform stream kullanarak gelen veri parçalarından JSON nesnelerini çıkarabilirsiniz.

Pratikte en sık karşılaşılan senaryo, büyük bir JSON dizisinin elemanlarını tek tek işlemektir. Örneğin bir API'den dışa aktarılan kullanıcı verileri genellikle büyük bir dizi içinde yer alır. Stream tabanlı yaklaşımla her kullanıcı kaydını bağımsız olarak okuyabilir, filtreleyebilir, dönüştürebilir ve veritabanına yazabilirsiniz.

Hata yönetimi, büyük JSON dosyalarında özellikle kritiktir. Dosyanın ortasında bir söz dizimi hatası varsa, tüm parse işlemi başarısız olabilir. Stream tabanlı yaklaşımda bu tür hataları daha erken tespit edebilir ve sorunlu bölümü atlayarak işleme devam edebilirsiniz. Bu özellik üretim ortamlarında büyük değer taşır çünkü tek bir bozuk kayıt yüzünden milyonlarca kaydın işlenmesinin durmasını istemezsiniz.

Bellek yönetimi konusunda bir ipucu daha: stream ile parse ederken oluşturduğunuz nesneleri işledikten sonra referanslarını serbest bırakmayı unutmayın. JavaScript'in çöp toplayıcısı referansı kaldırılmış nesneleri otomatik temizler ama büyük dizilerde veya closure'larda referanslar beklenmedik şekilde tutulabilir. Bu durum bellek sızıntısına yol açar ve stream yaklaşımının sağladığı avantajı ortadan kaldırır.

const fs = require('fs');
const { Transform, pipeline } = require('stream');
const path = require('path');

// Basit JSON dizi elemanı ayırıcı Transform stream
class JsonDiziAyirici extends Transform {
  constructor(secenekler = {}) {
    super({ readableObjectMode: true, ...secenekler });
    this.tampon = '';
    this.derinlik = 0;
    this.tirnakIcinde = false;
    this.kacisKarakteri = false;
    this.nesneBaslangic = -1;
    this.elemanSayisi = 0;
    this.diziBasladi = false;
  }

  _transform(parca, kodlama, geriCagir) {
    const veri = parca.toString();

    for (let i = 0; i < veri.length; i++) {
      const karakter = veri[i];

      // Kaçış karakteri kontrolü
      if (this.kacisKarakteri) {
        this.kacisKarakteri = false;
        if (this.nesneBaslangic >= 0) {
          this.tampon += karakter;
        }
        continue;
      }

      if (karakter === '\') {
        this.kacisKarakteri = true;
        if (this.nesneBaslangic >= 0) {
          this.tampon += karakter;
        }
        continue;
      }

      // Tırnak içi kontrolü
      if (karakter === '"') {
        this.tirnakIcinde = !this.tirnakIcinde;
        if (this.nesneBaslangic >= 0) {
          this.tampon += karakter;
        }
        continue;
      }

      if (this.tirnakIcinde) {
        if (this.nesneBaslangic >= 0) {
          this.tampon += karakter;
        }
        continue;
      }

      // Nesne sınır tespiti
      if (karakter === '{' || karakter === '[') {
        if (!this.diziBasladi && karakter === '[') {
          this.diziBasladi = true;
          continue;
        }
        if (this.derinlik === 0) {
          this.nesneBaslangic = i;
          this.tampon = karakter;
        } else {
          this.tampon += karakter;
        }
        this.derinlik++;
      } else if (karakter === '}' || karakter === ']') {
        if (this.derinlik > 0) {
          this.derinlik--;
          this.tampon += karakter;

          if (this.derinlik === 0 && this.tampon.length > 0) {
            try {
              const nesne = JSON.parse(this.tampon);
              this.elemanSayisi++;
              this.push(nesne);
            } catch (hata) {
              console.error('JSON parse hatası (eleman ' + this.elemanSayisi + '):', hata.message);
            }
            this.tampon = '';
            this.nesneBaslangic = -1;
          }
        }
      } else if (this.nesneBaslangic >= 0) {
        this.tampon += karakter;
      }
    }

    geriCagir();
  }

  _flush(geriCagir) {
    console.log('Toplam ayrıştırılan eleman: ' + this.elemanSayisi);
    geriCagir();
  }
}

// İşleyici Transform stream - her nesneyi bağımsız işler
class KayitIsleyici extends Transform {
  constructor(filtreFn, donusturFn) {
    super({ objectMode: true });
    this.filtreFn = filtreFn || (() => true);
    this.donusturFn = donusturFn || ((x) => x);
    this.islenenSayisi = 0;
    this.filtrelenenSayisi = 0;
  }

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

    if (this.islenenSayisi % 10000 === 0) {
      console.log(this.islenenSayisi + ' kayıt işlendi...');
    }

    if (this.filtreFn(kayit)) {
      const donusturulmus = this.donusturFn(kayit);
      this.push(JSON.stringify(donusturulmus) + '
');
    } else {
      this.filtrelenenSayisi++;
    }

    geriCagir();
  }

  _flush(geriCagir) {
    console.log('İşlenen: ' + this.islenenSayisi + ', Filtrelenen: ' + this.filtrelenenSayisi);
    geriCagir();
  }
}

// Ana işleme fonksiyonu
async function buyukJsonIsle(kaynakDosya, hedefDosya) {
  return new Promise((resolve, reject) => {
    const okuma = fs.createReadStream(kaynakDosya, { highWaterMark: 128 * 1024 });
    const ayirici = new JsonDiziAyirici();
    const isleyici = new KayitIsleyici(
      (kayit) => kayit.aktif !== false,
      (kayit) => ({ id: kayit.id, ad: kayit.ad, tarih: new Date().toISOString() })
    );
    const yazma = fs.createWriteStream(hedefDosya);

    pipeline(okuma, ayirici, isleyici, yazma, (hata) => {
      if (hata) {
        console.error('Pipeline hatası:', hata.message);
        reject(hata);
      } else {
        console.log('İşlem başarıyla tamamlandı!');
        resolve();
      }
    });
  });
}

// Kullanım örneği
buyukJsonIsle('./buyuk-veri.json', './islenenmis-veri.jsonl')
  .then(() => console.log('Tüm veriler işlendi.'))
  .catch((hata) => console.error('Hata oluştu:', hata.message));

Büyük JSON dosyalarını stream ile parse etmek, veri mühendisliğinin temel becerilerinden biridir. Custom Transform stream'ler yazarak kendi ihtiyaçlarınıza özel çözümler üretebilirsiniz. Bu yaklaşım sayesinde dosya boyutundan bağımsız, sabit bellek kullanımıyla veri işleme yapabilirsiniz.

 

[Best Studio]

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!