DataFrame ile ilk batch job — DE Zoomcamp (Gençler)

Batch İşin Anatomisi

“Batch” kelimesi İngilizce “parti, yığın” demektir. Veri mühendisliğinde batch processing = verinin birikmesini bekle, belirli bir zamanda (genelde gece veya saat başı) toplu olarak işle. Akış (streaming) ile zıt anlamlıdır.

Tipik batch pipeline:

Ingest: Veri bir yere düşer (CSV dosyası, Postgres tablosu, API response’u).
Validate: Tipler ve aralıklar kontrol edilir.
Transform: Temizle, zenginleştir, birleştir.
Aggregate: Grupla, topla, özetle.
Sink: Sonucu yaz (Parquet, BigQuery, rapor dosyası).

Bugünkü ders: adım 3-5’e odaklanıyoruz. Aracımız Polars (tek makine) ve kıyaslama için PySpark.

Polars Neden Varsayılan?

Bu kurs boyunca Polars kullanıyoruz. Pandas’tan farkları:

Sütun bazlı bellek: Bellek dostu (Pandas satır bazlı çalışır).
Lazy evaluation: pl.scan_csv() ile sorguyu planla, sonra collect() ile çalıştır. Optimizer arada gereksiz işleri atar.
Çok çekirdek: Paralel çalışır, modern CPU’ları tam kullanır.
Strict typing: Şema net. Veri kalitesi sorunları erkenden patlar.
5-30 kat hızlı: Pandas’a kıyasla tipik iş yüklerinde.

pip install polars pyarrow

Polars üç çekirdek kavram etrafında döner:

DataFrame: Tembel olabilen (LazyFrame) tablo.
Expression (pl.col): Bir sütun üzerinde hesap.
Context (select, with_columns, filter, group_by, sort): Expression’ları uyguladığın bağlam.

Uçtan Uca Bir Batch İş

Diyelim ki Elif’in e-ticaret verisi üç dosyada:

data/
├── orders.csv       (siparis_id, musteri_id, urun, fiyat, tarih)
├── customers.parquet (musteri_id, isim, sehir, kayit_tarihi)
└── products.parquet  (urun, kategori, stok_kodu)

Amacımız: “Mart 2026’da İstanbul’daki müşterilerin kategori bazlı toplam harcaması”.

1) Oku

Polars’ın scan_* API’si tembeldir: dosyayı henüz okumaz, sadece planı kurar.

import polars as pl

orders = pl.scan_csv("data/orders.csv")
customers = pl.scan_parquet("data/customers.parquet")
products = pl.scan_parquet("data/products.parquet")

2) Filtrele

filtered = (
    orders
    .filter(pl.col("tarih").is_between("2026-03-01", "2026-03-31"))
    .join(customers, on="musteri_id", how="inner")
    .join(products, on="urun", how="inner")
    .filter(pl.col("sehir") == "Istanbul")
)

3) Grupla ve Özetle

rapor = (
    filtered
    .group_by("kategori")
    .agg([
        pl.col("fiyat").sum().alias("toplam_harcama"),
        pl.col("fiyat").mean().alias("ortalama_sepet"),
        pl.len().alias("siparis_sayisi"),
        pl.col("musteri_id").n_unique().alias("tekil_musteri"),
    ])
    .sort("toplam_harcama", descending=True)
)

pl.len() satır sayısı, pl.col("musteri_id").n_unique() ise distinct count.

4) Topla ve Yaz

# Sorguyu çalıştır (tembel moddan çık)
sonuc = rapor.collect()
print(sonuc)

# Parquet olarak kaydet (sıkıştırılmış, sütun bazlı)
sonuc.write_parquet("out/istanbul_mart_2026.parquet")

collect() çağrısı query plan’ı çalıştırır. Polars burada otomatik olarak:

Gereksiz sütunları projection pruning ile atar.
Filtreleri erkene predicate pushdown yapar.
Join sırasını cost-based yeniden düzenler.

Eğer Pandas ile yazsaydık bunların hiçbiri olmazdı — bizim elle yazmamız gerekirdi.

Pandas’tan Polars’a Pratik Geçiş

Eğer daha önce Pandas öğrendiysen, dönüşüm hızlı:

import pandas as pd
import polars as pl

# Pandas -> Polars
pdf = pd.read_csv("eski.csv")
df = pl.from_pandas(pdf)

# Polars -> Pandas
df_pl = pl.read_csv("yeni.csv")
pdf_geri = df_pl.to_pandas()

API farkları:

Pandas	Polars
`df.groupby("k")`	`df.group_by("k")` (underscore)
`df["fiyat"].sum()`	`pl.col("fiyat").sum()`
`df.assign(toplam=...)`	`df.with_columns(toplam=pl.col("fiyat").sum())`
`df.query("fiyat > 100")`	`df.filter(pl.col("fiyat") > 100)`
`df.merge(b, on="id")`	`df.join(b, on="id")`

Kural: Yeni kod yazıyorsan doğrudan Polars başla. Pandas’a sadece dış kütüphaneler zorladığında (ör. scikit-learn) dön.

Spark ile Aynı İş (Karşılaştırma)

Aynı işi PySpark ile yazsaydık:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum, avg, countDistinct

spark = SparkSession.builder.appName("istanbul-rapor").getOrCreate()

orders = spark.read.csv("data/orders.csv", header=True, inferSchema=True)
customers = spark.read.parquet("data/customers.parquet")
products = spark.read.parquet("data/products.parquet")

rapor = (
    orders.filter(col("tarih").between("2026-03-01", "2026-03-31"))
    .join(customers, "musteri_id")
    .join(products, "urun")
    .filter(col("sehir") == "Istanbul")
    .groupBy("kategori")
    .agg(
        _sum("fiyat").alias("toplam_harcama"),
        avg("fiyat").alias("ortalama_sepet"),
        countDistinct("musteri_id").alias("tekil_musteri"),
    )
    .orderBy(col("toplam_harcama").desc())
)

rapor.write.parquet("out/istanbul_mart_2026.parquet")

Yapısal olarak aynı, ama:

Polars tek makinede ~2 saniyede biter (50 GB veri için 30 saniye).
PySpark 50 GB için 10 makinede ~1 dakikada biter.
Polars JVM yok, daha hafif; PySpark JVM başlatma maliyeti var.

Veri Kalitesi: schema Enforcement

Batch işin en önemli adımı veri kalitesi. Polars şema kabul eder, uymazsa hata fırlatır:

# Şema tanımla
schema = {
    "siparis_id": pl.Int64,
    "musteri_id": pl.Int64,
    "urun": pl.String,
    "fiyat": pl.Float64,
    "tarih": pl.Date,
}

# Okurken uygula
orders = pl.scan_csv("data/orders.csv", schema=schema, try_parse_dates=True)

try_parse_dates=True string tarihleri otomatik Date tipine çevirir. Eğer dosyada beklenmeyen bir tipli sütun varsa Polars hata verir — bu erken patlama iyidir, sessizce bozuk veri işlemekten iyidir.

Üretimde buna ek olarak great_expections veya pandera ile sütun düzeyinde kurallar koyarsın (ör. fiyat > 0, sehir izinli listede). Bu kursun ilerleyen derslerinde detaylı göreceğiz.

Streaming ile Batch’in Farkı

Polars ve PySpark batch içindir. Yani verinin tamamı elimizde, baştan sona işliyoruz. Streaming’de (sonraki modül) veri sürekli gelir ve sorgu sürekli çalışır.

Pratik ipucu: Çoğu “streaming” ihtiyacı aslında 5 dakikalık micro-batch ile karşılanır. Kafka’dan oku, her 5 dakikada bir Polars job çalıştır, BigQuery’ye yaz. Karmaşık streaming motorlarına gerek kalmaz.

Özet

Bugün öğrendik:

Batch processing = biriktir, toplu işle. Çoğu iş yükü için yeterli.
Polars = hızlı, tembel değerlendirmeli, çok çekirdekli DataFrame kütüphanesi.
group_by + agg kalbi — toplam, ortalama, distinct count tek seferde.
Lazy evaluation ile Polars planı optimize eder; Pandas’ta bunu sen yaparsın.
Pandas’tan Polars’a geçiş kolay; yeni kodda Polars.

Alıştırma

Yukarıdaki batch pipeline’ı 5 satırlık örnek dosyalarla kur, çalıştır.
pl.scan_csv yerine pl.read_csv dene. Plan ne zaman oluşuyor?
rapor.collect() öncesi rapor.explain() çağır. Optimizer planı gör.
Bir cümle yaz: “Hangi durumda batch yerine streaming seçerdim?”

Sıradaki

Batch işi yazdın. Şimdi bunu bulutta nasıl çalıştırırız öğreniyoruz →