polars, DuckDB ve Parquet: büyük veriye tek makinede

pandas yetmediğinde

Anadolu Market'in sipariş tablosu ilk yıl 200 bin satırdı; pandas ile saniyeler içinde gruplayıp raporladın (bunu pandas-groupby dersinde gördün). Üç yıl sonra tablo 400 milyon satır, CSV dosyası 60 GB. Dizüstü bilgisayarında 16 GB bellek var. pd.read_csv birkaç dakika diskle boğuştuktan sonra MemoryError verir ya da makine kilitlenir. Sorun senin kodunda değil, aracın tasarımında: pandas veriyi tamamen belleğe (eager, in-memory) yükler, çoğu işlemi tek çekirdekte çalıştırır ve ara sonuçlar için sık sık kopya üretir. Kabaca söylersek pandas'ın rahat çalıştığı sınır, verinin bellekte kapladığı alanın RAM'in beşte biri ile üçte biri arasında olmasıdır.

Bu derste o sınırı aşmanın dört yolunu öğreneceksin: veriyi parça parça okumak (chunksize), çok çekirdekli ve tembel çalışan polars, DataFrame'leri doğrudan SQL ile sorgulayan gömülü veritabanı DuckDB ve hepsinin ortak dili olan Apache Arrow / Parquet. Son olarak veri tek makineye sığmadığında devreye giren PySpark'ı tanıyacak ve "hangi boyutta hangi araç" kararını vereceksin.

1. Parça parça okuma: chunksize

En basit çözüm pandas'tan çıkmadan dosyayı dilimlerle okumaktır. chunksize verdiğinde read_csv tek bir DataFrame yerine bir yineleyici (iterator) döndürür; her adımda yalnızca o parça bellektedir. Toplam, sayım, en büyük gibi "birleştirilebilir" (associative) hesaplar için idealdir: her parçanın sonucunu bir sözlükte biriktirirsin.

import io
import pandas as pd

csv_metni = """siparis_id,sehir,tutar
1,İzmir,1500
2,Ankara,250
3,İzmir,90
4,Bursa,3200
5,Ankara,610
"""
ciro = {}
parca_sayisi = 0
for parca in pd.read_csv(io.StringIO(csv_metni), chunksize=2):
    parca_sayisi += 1
    for sehir, toplam in parca.groupby("sehir")["tutar"].sum().items():
        ciro[sehir] = ciro.get(sehir, 0) + int(toplam)
print("parça:", parca_sayisi)
print(dict(sorted(ciro.items())))

Gerçek hayatta io.StringIO yerine dosya yolu verirsin ve chunksize=1_000_000 gibi bir değer seçersin. Ayrıca yalnızca gereken sütunları usecols=["sehir", "tutar"] ile, uygun türlerle (dtype={"sehir": "category"}) okumak belleği yarı yarıya düşürebilir.

2. polars: hızlı ve tembel DataFrame

polars, Rust ile yazılmış, Apache Arrow bellek düzenini kullanan bir DataFrame kütüphanesidir. pandas'tan üç büyük farkı var: işlemleri tüm çekirdeklere dağıtır, indeks (index) kavramı yoktur ve en önemlisi ifade (expression) tabanlı bir dili vardır. pl.col("tutar") bir sütunu değil, "tutar sütunu üzerinde yapılacak hesabın tarifini" temsil eder; bu tarifleri select, with_columns, filter ve agg içinde kullanırsın.

import polars as pl

siparisler = pl.DataFrame({
    "siparis_id": [1, 2, 3, 4, 5, 6],
    "sehir": ["İzmir", "Ankara", "İzmir", "Bursa", "Ankara", "İzmir"],
    "tutar": [1500, 250, 90, 3200, 610, 720],
})
print(siparisler.shape, siparisler.columns)

buyukler = siparisler.filter(pl.col("tutar") >= 500)
print(buyukler["siparis_id"].to_list())

kdvli = siparisler.with_columns(
    (pl.col("tutar") * 1.2).round(2).alias("kdvli"),
    (pl.col("tutar") >= 1000).alias("buyuk_siparis"),
)
print(kdvli.columns)

ozet = (
    siparisler.group_by("sehir")
    .agg(pl.col("tutar").sum().alias("ciro"), pl.len().alias("adet"))
    .sort("ciro", descending=True)
)
print(ozet.rows())

Okurken şuna dikkat et: hiçbir yerde df["yeni"] = ... gibi yerinde atama yok. polars'ta her işlem yeni bir DataFrame döndürür; zincirleme yazarsın. with_columns sütun ekler ya da değiştirir, select yalnızca istediğin sütunları bırakır, alias sonucun adını verir. group_by çıktısının satır sırası garanti değildir; raporda her zaman sort ekle.

Lazy API: önce plan, sonra çalıştır

polars'ın asıl gücü tembel (lazy) moddur. df.lazy() ya da doğrudan dosyadan pl.scan_csv(...) / pl.scan_parquet(...) bir LazyFrame döndürür. Üzerine yazdığın filter, group_by, select hemen çalışmaz; polars bir sorgu planı (query plan) kurar. collect() dediğinde planı optimize eder: süzmeyi okumanın içine iter (predicate pushdown), kullanılmayan sütunları hiç okumaz (projection pushdown) ve gerekiyorsa veriyi akış hâlinde (streaming) işler. Sonuç: 60 GB'lık dosyanın yalnızca iki sütununu ve ilgili satırlarını okursun.

Tarayıcıdaki polars derlemesi dosyadan tembel okumayı (scan_csv, scan_parquet) ve polars'ın kendi Parquet okuyucusunu desteklemiyor. Bu sayfadaki çalışan örnekler aynı planı pl.read_csv(...).lazy() ile kuruyor; kendi bilgisayarında pl.scan_csv(...) yazman yeterli, gerisi aynı.

import polars as pl

with open("siparisler.csv", "w", encoding="utf-8") as f:
    f.write("gun,sehir,tutar,iade\n")
    f.write("2025-03-01,İzmir,1500,0\n2025-03-01,Ankara,250,1\n")
    f.write("2025-03-02,İzmir,90,0\n2025-03-02,Bursa,3200,0\n")

plan = (
    pl.read_csv("siparisler.csv").lazy()   # bilgisayarında: pl.scan_csv("siparisler.csv")
    .filter(pl.col("iade") == 0)
    .group_by("gun")
    .agg(pl.col("tutar").sum().alias("ciro"))
    .sort("gun")
)
print(type(plan).__name__)          # henüz hiçbir şey okunmadı
print(plan.collect().rows())        # şimdi çalıştı

3. DuckDB: DataFrame'e SQL

Ekipte herkes SQL biliyorsa (Kesit SQL'den hatırlarsın) DuckDB en kısa yoldur. Kurulum gerektirmeyen, süreç içinde çalışan (in-process) bir analitik veritabanıdır; SQLite'ın analitik kardeşi gibi düşün. En sevilen özelliği: Python değişkenindeki bir pandas ya da polars DataFrame'ini tablo adıymış gibi sorgulayabilirsin. Sonucu .fetchall() ile demet listesi, .df() ile pandas, .pl() ile polars olarak alırsın.

import duckdb
import polars as pl

siparisler = pl.DataFrame({
    "musteri": ["Ayşe", "Mert", "Ayşe", "Zeynep", "Mert", "Ayşe"],
    "sehir": ["İzmir", "Ankara", "İzmir", "Bursa", "Ankara", "İzmir"],
    "tutar": [1500, 250, 90, 3200, 610, 720],
})
sorgu = """
    SELECT musteri, SUM(tutar) AS toplam, COUNT(*) AS adet
    FROM siparisler
    WHERE sehir <> 'Bursa'
    GROUP BY musteri
    ORDER BY toplam DESC
"""
print(duckdb.sql(sorgu).fetchall())
pdf = duckdb.sql(sorgu).df()          # pandas DataFrame
print(list(pdf.columns), len(pdf))

DuckDB de sütun tabanlıdır, tüm çekirdekleri kullanır ve bellekten büyük veride diske taşarak çalışabilir. Üstelik dosyaları doğrudan sorgular: SELECT * FROM 'satislar/*.parquet' yazman yeterli; yükleme adımı yok.

4. Arrow ve Parquet: ortak dil

Apache Arrow, sütun tabanlı bir bellek biçimidir; polars, DuckDB ve pandas 2+ bu biçimi konuşur, böylece araçlar arasında veri kopyalanmadan el değiştirir. Parquet ise Arrow'un diskteki karşılığı gibidir: sütun tabanlı, şemalı (her sütunun türü dosyada yazar) ve sıkıştırılmış bir dosya biçimi. CSV'ye göre genellikle birkaç kat küçük ve okuması çok daha hızlıdır; çünkü yalnızca istenen sütunlar okunur ve her blokta min/max istatistikleri sayesinde ilgisiz bloklar atlanır.

import pyarrow as pa
import pyarrow.parquet as pq
import polars as pl
import duckdb

tablo = pa.table({
    "musteri_id": [101, 102, 103, 104],
    "segment": ["Bireysel", "KOBİ", "Bireysel", "Kurumsal"],
    "bakiye": [12500.0, 98000.5, 430.0, 1250000.0],
})
pq.write_table(tablo, "musteriler.parquet")

sadece_iki = pq.read_table("musteriler.parquet", columns=["musteri_id", "bakiye"])
print(sadece_iki.column_names, sadece_iki.num_rows)
print(pq.read_schema("musteriler.parquet").field("bakiye").type)

df = pl.from_arrow(pq.read_table("musteriler.parquet"))   # bilgisayarında: pl.read_parquet(...)
print(df.filter(pl.col("segment") == "Bireysel")["musteri_id"].to_list())
print(duckdb.sql("SELECT segment, COUNT(*) FROM 'musteriler.parquet' GROUP BY segment ORDER BY segment").fetchall())

5. PySpark: veri tek makineye sığmadığında

Veri terabaytlar seviyesine çıktığında ya da sürekli büyüyen bir veri gölünde (data lake) onlarca ekip aynı veriye eriştiğinde tek makine yetmez. Apache Spark işi bir küme (cluster) üzerindeki birçok makineye dağıtır; PySpark bunun Python arayüzüdür. API'si polars'a benzer: tembel DataFrame, ifadeler ve sonuç için bir eylem (action).

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("anadolu-market").getOrCreate()
df = spark.read.parquet("s3://anadolu-market/siparisler/")   # yüzlerce dosya
ozet = (
    df.filter(F.col("iade") == 0)
      .groupBy("sehir")
      .agg(F.sum("tutar").alias("ciro"), F.count("*").alias("adet"))
      .orderBy(F.desc("ciro"))
)
ozet.show(10)          # eylem: plan burada kümede çalışır
spark.stop()

Hangi boyutta hangi araç?

Bu sınırlar kesin değildir; verinin bellekte kapladığı alan sütun türlerine göre CSV boyutundan büyük ya da küçük olabilir. Pratik yol: önce küçük bir örnekle ölç (performans dersindeki gibi), sonra karar ver.

gol/
├── ay=2024-06/data_0.parquet
├── ay=2024-07/data_0.parquet
└── ay=2024-08/data_0.parquet

import veri, duckdb, pathlib, shutil
yol = veri.yol("pedal_izmir", "bk_trips")
shutil.rmtree("gol", ignore_errors=True)
duckdb.sql(f"""
    COPY (SELECT *, strftime(started_at::TIMESTAMP, '%Y-%m') AS ay FROM read_csv_auto('{yol}'))
    TO 'gol' (FORMAT parquet, PARTITION_BY (ay))
""")
print(sorted(str(p) for p in pathlib.Path("gol").rglob("*.parquet")))
print(duckdb.sql("""
    SELECT ay, COUNT(*) AS yolculuk, ROUND(SUM(fare), 2) AS ciro
    FROM read_parquet('gol/**/*.parquet', hive_partitioning = true)
    GROUP BY ay ORDER BY ay
""").fetchall())

import pyarrow.parquet as pq
pq.write_to_dataset(tablo, root_path="gol", partition_cols=["ay"])      # pyarrow

df.write_parquet("gol", partition_by="ay")                              # polars (yeni sürümler)
pl.scan_parquet("gol/**/*.parquet", hive_partitioning=True).filter(pl.col("ay") == "2024-07")

EXPLAIN ANALYZE
SELECT COUNT(*) FROM read_parquet('gol/**/*.parquet', hive_partitioning = true)
WHERE ay = '2024-07';

│ File Filters: ay = '2024-07'    │
│ Scanning Files: 1/3             │
│ Total Files Read: 1             │

import polars as pl

df = pl.DataFrame({
    "siparis_id": [1, 2, 3, 4, 5, 6],
    "sehir": ["İzmir", "İzmir", "İzmir", "Ankara", "Ankara", "Bursa"],
    "tutar": [100, 120, 900, 300, 320, 50],
})
isaretli = df.with_columns(
    pl.col("tutar").mean().over("sehir").round(1).alias("sehir_ort")
).with_columns(
    (pl.col("tutar") > 2 * pl.col("sehir_ort")).alias("supheli")
)
print(isaretli.filter(pl.col("supheli"))["siparis_id"].to_list())
print(isaretli["sehir_ort"].to_list())