CSV, JSON ve Parquet Arasında Dönüşüm: Bellek Taşırmadan Format Hattı

CSV, JSON ve Parquet arasında dönüşüm, veri işinde en sık tekrarlanan ama en çok bellek taşıran adımlardan biridir. Dışa aktarılan bir CSV birkaç yüz megabayta ulaştığında alışıldık yöntem, dosyayı tek seferde bir DataFrame'e yükleyip başka biçimde kaydetmektir; bu yaklaşım 1 GB'lık bir dosyada makinenin belleğini doldurur ve betik ya yavaşlar ya da sessizce ölür.

Oysa üç biçim de akış halinde işlenebilir. CSV satır satır okunur, JSON satırlı biçime (JSONL) çevrilince satır satır yazılır, Parquet ise satır gruplarına bölünmüş sütunlu bir dosya olduğundan parça parça yazılmaya zaten uygundur. Belleğe her an yalnız bir parça sığar; dosya ne kadar büyürse büyüsün betiğin bellek ihtiyacı sabit kalır.

Bu yazıda parça parça okuma, JSONL akışı, Parquet'e yazarken şema ve tip sabitleme ve sıkıştırma seçimini sırayla ele alıyorum. İki kod örneği var: sabit bellekle CSV'yi Parquet'e çeviren betik ve dönüşüm sonrası satır sayısıyla sütun toplamlarını doğrulayan betik.

Parça parça okuma ve sabit bellek

Parça parça okuma, dosyanın tamamını değil belirli boyutta bir bloğunu belleğe alıp işleyen, sonra bir sonraki bloğa geçen okuma biçimidir. Bellekte aynı anda yalnız bir blok bulunduğundan tüketim dosya boyutundan bağımsız hale gelir. pandas'ta chunksize parametresi, pyarrow'da open_csv ile dönen akış okuyucu bu işi yapar.

Blok boyutu iki şeyi dengeler. Küçük blok az bellek ister ama daha çok yazma turu ve daha çok küçük satır grubu üretir; büyük blok daha az tur atar ama bellek zirvesini yükseltir. Başlangıç için birkaç on megabaytlık blok çoğu makinede rahat çalışır; doğru değer, betik çalışırken bellek kullanımı ölçülerek bulunur.

Bellek zirvesi şöyle ölçülür: Linux'ta /usr/bin/time -v python3 donustur.py komutunun çıktısındaki en yüksek yerleşik küme boyutu satırı, Python içinde ise tracemalloc modülünün zirve değeri. Blok boyutunu değiştirip aynı ölçümü tekrarlamak, dosya büyüse de zirvenin sabit kaldığını gösterir.

JSON'u satırlı biçime alarak akıtmak

JSONL, her satırın kendi başına geçerli bir JSON nesnesi olduğu metin biçimidir ve büyük veri için tek bir devasa JSON dizisinden çok daha uygundur. Standart json.load bir diziyi ancak tamamını belleğe alarak ayrıştırır; JSONL'de ise dosya satır satır okunur ve her satır bağımsız çözülür. API'lerden gelen kayıtlar ve log dışa aktarımları çoğu zaman zaten bu biçimdedir.

Elinizdeki dosya tek bir dizi ise önce JSONL'e çevirmek en güvenli yoldur. Bu dönüşüm için standart kütüphanedeki json yeterli olmaz; ijson gibi artımlı bir ayrıştırıcı diziyi öğe öğe verir ve her öğe tek satır olarak yazılır. Sonrasında JSONL dosyası, tıpkı CSV gibi, blok blok Parquet'e alınır; pyarrow'un pyarrow.json.read_json fonksiyonu bu biçimi doğrudan tanır.

İç içe alanlar dönüşümün en çok takıldığı yerdir. Parquet iç içe yapıları destekler ama tablo araçlarının çoğu düz sütun bekler; adres.il gibi alanları noktalı sütun adına düzleştirmek, listeleri ise ayrı bir tabloya çıkarmak sonraki sorguları kolaylaştırır.

Parquet'e yazarken şema ve tip sabitleme

Şema sabitleme, her sütunun tipini dosya yazılmadan önce açıkça bildirmek ve okuyucunun bloktan bloğa tip tahmini yapmasını engellemektir. CSV tip taşımaz; okuyucu ilk bloğa bakıp tahmin yürütür ve tahmin sonraki blokta çürüyebilir. Sıfırla başlayan müşteri numaralarının tam sayı sanılıp baştaki sıfırların düşmesi bu hatanın en bilinen örneğidir.

Aşağıdaki betik pyarrow'un akış okuyucusuyla CSV'yi bloklar halinde okur, sütun tiplerini açıkça verir ve her bloğu aynı Parquet yazıcıya satır grubu olarak ekler. Bellekte aynı anda yalnız bir blok bulunur. Boş değerlerin nasıl yazıldığını bilmek önemlidir; null_values listesi bu yüzden açıkça verilir.

import sys

import pyarrow as pa
import pyarrow.csv as pcsv
import pyarrow.parquet as pq

SEMA = {
    "musteri_no": pa.string(),
    "tarih": pa.date32(),
    "urun": pa.string(),
    "adet": pa.int32(),
    "tutar": pa.decimal128(12, 2),
}
BLOK_BAYT = 32 * 1024 * 1024


def csv_to_parquet(kaynak, hedef, sikistirma="zstd"):
    okuma = pcsv.ReadOptions(block_size=BLOK_BAYT)
    donusum = pcsv.ConvertOptions(
        column_types=SEMA,
        null_values=["", "NA", "NULL"],
        strings_can_be_null=True,
    )
    satir = 0
    with pcsv.open_csv(kaynak, read_options=okuma, convert_options=donusum) as okuyucu:
        with pq.ParquetWriter(hedef, okuyucu.schema, compression=sikistirma) as yazici:
            for blok in okuyucu:
                yazici.write_batch(blok)
                satir += blok.num_rows
    return satir


if __name__ == "__main__":
    kaynak, hedef = sys.argv[1], sys.argv[2]
    print(csv_to_parquet(kaynak, hedef), "satır yazıldı:", hedef)

Tarih sütunu date32 olarak bildirildiğinde yalnız ISO biçimindeki değerler ayrıştırılır; gün.ay.yıl gibi yerel biçimler için timestamp_parsers seçeneğine desen eklenir. Türkçe ondalık ayracı ve karakter kodlaması sorunları da bu aşamada çıkar; tuzakların dökümü veri normalizasyonu yazısında var.

Sıkıştırma ve satır grubu seçimi

Sıkıştırma seçimi, Parquet dosyasının disk boyutu ile okuma ve yazma hızı arasındaki dengeyi belirleyen ayardır. Snappy hızlı ama az sıkıştırır, gzip küçük dosya üretir ama yavaştır, zstd ikisinin arasında durur ve çoğu araç tarafından desteklenir. Dosya bir kez yazılıp çok kez okunacaksa okuma hızı, arşivlenecekse boyut öncelik alır.

Seçim tahminle değil ölçümle yapılır. Aynı CSV üç sıkıştırmayla ayrı ayrı Parquet'e çevrilir, her çıktının boyutu os.path.getsize ile, yazma ve okuma süresi time.perf_counter ile kaydedilir; kendi verinizde hangi ayarın kazandığı bu küçük tablodan okunur. Sütun içeriği sonucu değiştirir: tekrarlı metin sütunları her sıkıştırmada küçülür, rastgele kimlikler neredeyse hiç küçülmez.

Satır grubu boyutu da okuma davranışını etkiler. Sorgu araçları satır grubu başlığındaki en küçük ve en büyük değerlere bakarak ilgisiz grupları atlar; bu yüzden tarih sütununa göre sıralı yazılmış bir dosya, tarih filtresiyle çok daha az veri okur. Blok boyutu satır grubunu belirlediğinden bu iki ayar birlikte düşünülür.

Dönüşüm sonrası doğrulama

Dönüşüm doğrulaması, kaynak ve hedef dosyanın aynı veriyi taşıdığını satır sayısı ve sütun toplamlarıyla kanıtlayan kontroldür. Betik hatasız bitmiş olabilir ama boş değer kuralı bir sütunu sessizce sıfırlamış ya da bir blok atlanmış olabilir. Satır sayısı eşitliği ve sayısal sütunların toplam eşitliği bu hataların büyük bölümünü yakalar.

Aşağıdaki betik kaynağı yeniden akış halinde okuyup satır sayısını ve seçilen sütunların toplamını hesaplar, ardından Parquet dosyasının üst verisindeki satır sayısını ve aynı sütunların toplamını karşılaştırır. Parquet üst verisi dosyayı okumadan satır sayısını verir; toplam için ise yalnız ilgili sütunlar okunur.

import sys

import pyarrow.compute as pc
import pyarrow.csv as pcsv
import pyarrow.parquet as pq

TOPLANACAK = ["adet", "tutar"]


def kaynak_ozeti(csv_yolu):
    satir, toplam = 0, {ad: 0 for ad in TOPLANACAK}
    with pcsv.open_csv(csv_yolu) as okuyucu:
        for blok in okuyucu:
            satir += blok.num_rows
            for ad in TOPLANACAK:
                toplam[ad] += pc.sum(blok.column(ad)).as_py() or 0
    return satir, toplam


def hedef_ozeti(parquet_yolu):
    dosya = pq.ParquetFile(parquet_yolu)
    tablo = dosya.read(columns=TOPLANACAK)
    toplam = {ad: pc.sum(tablo.column(ad)).as_py() or 0 for ad in TOPLANACAK}
    return dosya.metadata.num_rows, toplam


def dogrula(csv_yolu, parquet_yolu):
    k_satir, k_toplam = kaynak_ozeti(csv_yolu)
    h_satir, h_toplam = hedef_ozeti(parquet_yolu)
    sorunlar = []
    if k_satir != h_satir:
        sorunlar.append("satır sayısı: kaynak %d, hedef %d" % (k_satir, h_satir))
    for ad in TOPLANACAK:
        if float(k_toplam[ad]) != float(h_toplam[ad]):
            sorunlar.append("%s toplamı: kaynak %s, hedef %s" % (ad, k_toplam[ad], h_toplam[ad]))
    return sorunlar


if __name__ == "__main__":
    sorunlar = dogrula(sys.argv[1], sys.argv[2])
    if sorunlar:
        print("DOĞRULAMA BAŞARISIZ:", "; ".join(sorunlar))
        sys.exit(1)
    print("doğrulandı: satır sayısı ve sütun toplamları eşit")

Kaynak tarafında tip bildirilmediği için pyarrow toplanan sütunları tahminle sayıya çevirir; tahmin başarısız olursa toplama hata verir ve bu da bir bulgudur. Doğrulama betiği dönüşümün hemen ardından, aynı zincirde çalıştırılır; çıkış kodu sıfırdan farklıysa hedef dosya silinir ve iş başarısız sayılır. Bu zincirin kilit ve yeniden çalıştırma düzeni için zamanlanmış iş tasarımı yazısına bakılabilir.

Sonuç: dönüşüm hattı ölçülerek kurulur

Format dönüşümü bir bellek sorunu değil, bir akış tasarımı sorunudur. CSV blok blok okunur, JSON satırlı biçime alınır, Parquet açıkça bildirilmiş şemayla satır grupları halinde yazılır ve sıkıştırma kendi veride ölçülerek seçilir. Dönüşümün sonunda satır sayısı ve sütun toplamları karşılaştırılmadan iş bitmiş sayılmaz.

İlk adım olarak elinizdeki en büyük CSV'yi seçin, sütun tiplerini yazın ve akış betiğini bellek ölçümüyle çalıştırın. Ardından doğrulama betiğini aynı zincire ekleyin; tip tahmininin nerede çürüdüğü ilk çalışmada ortaya çıkar. Sıkıştırma karşılaştırmasını en sona bırakın, çünkü o karar ancak doğru tiplenmiş bir dosya üzerinde anlamlıdır.

Büyük dışa aktarımlarınızı sabit bellekle işleyen bir dönüşüm hattı kurmak için iletişim bölümünden yazabilirsiniz.