PySpark ile dağıtık veri işleme: partition, shuffle ve Spark SQL

Neden önemli? Veri tek makineye sığmadığında

Pedal İzmir'in bisiklet paylaşım sistemi üç yaz ayında 116.517 yolculuk üretti. Bu boyut bir dizüstü bilgisayar için çocuk oyuncağı: polars ya da DuckDB saniyeler içinde özetler (bunu polars, DuckDB ve Parquet: büyük veriye tek makinede dersinde gördün). Şimdi aynı şirketin 30 şehre yayıldığını, her bisikletin 5 saniyede bir GPS konumu gönderdiğini düşün: günde milyarlarca satır, yılda yüzlerce terabayt. Tek makinenin belleği, diski ve işlemcisi yetmez. Çözüm işi parçalara bölüp çok sayıda makineye dağıtmaktır. Apache Spark bu işin endüstri standardıdır; Databricks, AWS EMR, Google Dataproc ve Azure Synapse'in kalbinde o çalışır.

Spark, Java sanal makinesi (JVM) ve bir küme ister; tarayıcıda çalışmaz. Bu yüzden derste PySpark kodunu gri kutularda okuyacak, Spark'ın arka planda yaptığı işi ise saf Python, polars ve DuckDB ile kendin kuracaksın: bölümleme (partitioning), map-reduce, shuffle, broadcast join, salting ve pencere fonksiyonları (window function). Mülakatlarda sorulan da tam olarak bu iç işleyiştir; API'yi bir öğleden sonrada öğrenirsin, ama "bu iş neden 40 dakika sürüyor?" sorusunu iç işleyişi bilen cevaplar.

Dağıtık hesaplamanın parçaları: sürücü, yürütücü, partition

Bir Spark uygulamasında tek bir sürücü (driver) programı ve birçok yürütücü (executor) süreci vardır. Sürücü senin Python kodunu çalıştırır, planı çıkarır ve işi küçük görevlere (task) bölüp yürütücülere dağıtır. Veri, partition denen parçalara ayrılmıştır; her görev tek bir partition üzerinde çalışır. 200 partition ve 50 çekirdek varsa görevler dört dalgada biter.

Bölümlemenin en yaygın yolu hash bölümlemedir: anahtarın özeti (hash) partition sayısına bölünür, kalan partition numarasını verir. Aynı anahtar her zaman aynı partition'a düşer. Aşağıda Pedal İzmir yolculuklarını başlangıç istasyonuna göre 4 partition'a dağıtıyoruz (tamsayı anahtarda % n yeterli):

import veri

trips = veri.kayitlar("pedal_izmir", "bk_trips")
print(len(trips), "yolculuk")

n = 4
bolumler = [[] for _ in range(n)]
for t in trips:
    bolumler[t["start_station"] % n].append(t)

for i, b in enumerate(bolumler):
    print(f"partition {i}: {len(b):>6} kayıt")

SparkSession, DataFrame API ve tembel değerlendirme

PySpark'ta her şey bir SparkSession ile başlar. DataFrame API'si pandas'tan çok polars'ın lazy API'sine benzer: filter, select, groupBy gibi dönüşümler (transformation) hiçbir şey hesaplamaz, yalnız plan kurar. count, show, collect, write gibi eylemler (action) gelince Catalyst iyileştiricisi planı optimize eder ve küme çalışmaya başlar.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("pedal-izmir").getOrCreate()

trips = spark.read.csv("s3://pedal/bk_trips/", header=True, inferSchema=True)
stations = spark.read.parquet("s3://pedal/bk_stations/")

gunluk = (trips
          .filter(F.col("distance_km") > 0)                 # dönüşüm: plan
          .withColumn("gun", F.to_date("started_at"))       # dönüşüm: plan
          .groupBy("gun")                                   # dönüşüm: plan (geniş)
          .agg(F.count("*").alias("yolculuk"),
               F.round(F.avg("fare"), 2).alias("ort_ucret")))

gunluk.explain()          # fiziksel planı yazdırır, veri okumaz
gunluk.show(5)            # EYLEM: iş burada başlar

Aynı tembel mantığı tarayıcıda polars ile görebilirsin: .lazy() (bilgisayarında doğrudan dosyadan scan_csv) plan kurar, collect çalıştırır.

import polars as pl

plan = (pl.read_csv(veri.yol("pedal_izmir", "bk_trips")).lazy()   # bilgisayarında: pl.scan_csv(...)
        .filter(pl.col("distance_km") > 0)
        .with_columns(pl.col("started_at").str.slice(0, 7).alias("ay"))
        .group_by("ay")
        .agg(pl.len().alias("yolculuk"), pl.col("fare").mean().round(2).alias("ort_ucret"))
        .sort("ay"))
print(type(plan).__name__)      # henüz veri yok, yalnız plan
print(plan.collect())           # eylem

Dar ve geniş dönüşümler: shuffle'ın bedeli

Dar dönüşüm (narrow) her çıktı partition'ının yalnız bir girdi partition'ına bağlı olduğu işlemdir: filter, select, withColumn. Ağ trafiği yoktur, her yürütücü kendi parçasını işler. Geniş dönüşüm (wide) ise aynı anahtara sahip kayıtların aynı yere toplanmasını gerektirir: groupBy, join, distinct, orderBy. Bu toplanmaya shuffle denir: veriler diske yazılır, ağdan taşınır, yeniden okunur. Spark işlerinin yavaşlığının bir numaralı sebebi shuffle'dır.

Spark'ın groupBy'ı akıllıdır: shuffle'dan önce her partition kendi içinde kısmi toplam çıkarır (map-side combine), ağdan yalnız bu özetler gider. Aşağıda dosyanın ardışık parçalara bölündüğü durumu (HDFS ya da S3 blokları gibi) taklit ediyoruz ve combiner'ın ağ trafiğini ne kadar düşürdüğünü sayıyoruz:

from collections import Counter

def parcala(kayitlar, n):
    boy = -(-len(kayitlar) // n)              # yukarı yuvarlayan bölme
    return [kayitlar[i * boy:(i + 1) * boy] for i in range(n)]

parcalar = parcala(trips, 4)

# MAP + COMBINE: her parça kendi içinde sayar
kismi = [Counter(t["start_station"] for t in p) for p in parcalar]
print("combiner olmasa ağdan giden kayıt:", sum(len(p) for p in parcalar))
print("combiner ile ağdan giden kayıt   :", sum(len(k) for k in kismi))

# SHUFFLE + REDUCE: aynı anahtarlar birleşir
toplam = Counter()
for k in kismi:
    toplam.update(k)
print("en yoğun 3 istasyon:", toplam.most_common(3))

116 binden 248'e: combiner sayesinde shuffle'a binlerce kat daha az kayıt girer. Ortalama gibi birleştirilebilir olmayan ölçülerde ise kısmi sonuç olarak (toplam, adet) çifti taşınır; ortalamaların ortalaması yanlış sonuç verir (alıştırmalarda bu hatayı avlayacaksın).

Join stratejileri, veri çarpıklığı ve salting

İki büyük tabloyu birleştirmek için Spark varsayılan olarak sort-merge join kullanır: iki taraf da anahtara göre shuffle edilip sıralanır, sonra fermuar gibi birleştirilir. Taraflardan biri küçükse (varsayılan eşik 10 MB, spark.sql.autoBroadcastJoinThreshold) broadcast hash join seçilir: küçük tablo her yürütücüye kopyalanır, büyük tablo yerinden kıpırdamaz. 62 satırlık istasyon tablosu bunun ders kitabı örneğidir.

from pyspark.sql.functions import broadcast

zengin = trips.join(broadcast(stations),
                    trips.start_station == stations.station_id, "left")
zengin.groupBy("district").count().orderBy(F.desc("count")).show()

stations = veri.kayitlar("pedal_izmir", "bk_stations")
ilce = {s["station_id"]: s["district"] for s in stations}   # her yürütücüye giden küçük sözlük

sayac = Counter(ilce[t["start_station"]] for t in trips)     # büyük tablo yerinde kalır
for ad, n in sayac.most_common(3):
    print(ad, n)

Veri çarpıklığı (skew), bazı anahtarların diğerlerinden çok daha sık görülmesidir. Alsancak Çarşı istasyonu tek başına 4.257 yolculukla öndedir; gerçek hayatta bu oran 1'e 1000 olabilir (ör. "bilinmiyor" müşteri kimliği). Hash bölümlemede o anahtarın bütün kayıtları tek göreve düşer; 199 görev 10 saniyede biterken bir görev 40 dakika sürer. Spark UI'da bu durum "görev sürelerinin max değeri medyanın çok üstünde" diye görünür.

Çözümlerden biri salting (tuzlama): sıcak anahtara yapay bir son ek eklenir (12#0, 12#1...), böylece yük birkaç göreve yayılır. İlk aşamada tuzlu anahtarlarla kısmi toplam alınır, ikinci aşamada tuz atılıp yeniden toplanır. Spark 3'ün AQE (adaptive query execution) özelliği join'lerdeki çarpıklığı çoğu zaman kendisi böler (spark.sql.adaptive.skewJoin.enabled), ama groupBy çarpıklığında ve eski sürümlerde salting hâlâ elle yapılır.

sade = Counter(t["start_station"] for t in trips)
sicak = {k for k, _ in sade.most_common(2)}
tuzlu = Counter((t["start_station"], t["trip_id"] % 4 if t["start_station"] in sicak else 0) for t in trips)
print("tuzsuz en büyük grup:", max(sade.values()))
print("tuzlu en büyük grup :", max(tuzlu.values()))

Spark SQL ve pencere fonksiyonları

DataFrame'i geçici görünüm olarak kaydedip SQL yazabilirsin; Spark SQL ve DataFrame API aynı iyileştiriciye gider, performans farkı yoktur. Pencere fonksiyonları (ROW_NUMBER, RANK, LAG, kayan toplam) da aynıdır. PARTITION BY ifadesi Spark'ta gerçekten bir shuffle'a dönüşür: aynı ilçenin satırları aynı göreve toplanır.

trips.createOrReplaceTempView("trips")
stations.createOrReplaceTempView("stations")
spark.sql("""
    WITH s AS (
        SELECT st.district, st.name, COUNT(*) AS n
        FROM trips t JOIN stations st ON t.start_station = st.station_id
        GROUP BY st.district, st.name)
    SELECT * FROM (
        SELECT *, ROW_NUMBER() OVER (PARTITION BY district ORDER BY n DESC) AS sira
        FROM s) WHERE sira = 1
""").show()

Aynı sorguyu DuckDB ile tarayıcıda çalıştırabilirsin; SQL neredeyse harfi harfine aynı:

import duckdb

t_yol = veri.yol("pedal_izmir", "bk_trips")
s_yol = veri.yol("pedal_izmir", "bk_stations")
sorgu = f"""
WITH s AS (
    SELECT st.district, st.name, COUNT(*) AS n
    FROM read_csv('{t_yol}') t JOIN read_csv('{s_yol}') st ON t.start_station = st.station_id
    GROUP BY st.district, st.name)
SELECT district, name, n FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY district ORDER BY n DESC, name) AS sira FROM s)
WHERE sira = 1 ORDER BY n DESC LIMIT 3
"""
for satir in duckdb.sql(sorgu).fetchall():
    print(satir)

Parquet, partitionBy, Delta ve Iceberg

Spark'ın doğal dosya biçimi Parquet'tir: sütunsal, sıkıştırılmış ve şemalı. partitionBy ile yazınca veri hive tarzı klasörlere ayrılır (ay=2024-06/). Sorgu WHERE ay = '2024-07' içerirse Spark öteki klasörleri hiç açmaz; buna bölüm budama (partition pruning) denir. Bölüm sütunu, filtrede sık kullanılan ve az sayıda farklı değeri olan bir sütun olmalıdır: gün ya da ay iyi, müşteri kimliği felakettir (milyonlarca küçük dosya).

(trips.withColumn("ay", F.substring("started_at", 1, 7))
      .repartition("ay")                       # her ay tek görevde: az ve büyük dosya
      .write.mode("overwrite")
      .partitionBy("ay")
      .parquet("s3://pedal/lake/trips/"))

# Delta Lake: Parquet + işlem günlüğü (ACID, MERGE, zaman yolculuğu)
trips.write.format("delta").mode("append").save("s3://pedal/delta/trips")
spark.read.format("delta").option("versionAsOf", 3).load("s3://pedal/delta/trips")

UDF'nin bedeli ve pandas UDF

Spark'ın hazır fonksiyonları (F.when, F.regexp_extract...) JVM içinde, Catalyst'in optimize ettiği kodla çalışır. Python ile yazdığın sıradan bir UDF ise her satırı JVM'den Python sürecine taşır, sonucu geri getirir: serileştirme maliyeti işi 10 kat yavaşlatabilir ve iyileştirici fonksiyonun içini göremez. Önce hazır fonksiyon ara; yoksa Arrow ile toplu (vektörel) çalışan pandas UDF kullan.

from pyspark.sql.functions import pandas_udf
import pandas as pd

@F.udf("double")                       # yavaş: satır satır
def kdv_ekle(tutar):
    return tutar * 1.2

@pandas_udf("double")                  # hızlı: Arrow ile toplu
def kdv_ekle_vek(tutar: pd.Series) -> pd.Series:
    return tutar * 1.2

trips.select(F.col("fare") * 1.2)      # en hızlı: hazır ifade, UDF yok

Ne zaman Spark gereksizdir?