Bir Spark işi çalıştırdığınızda aslında neler olur?

Giriş

Veri kümeleri büyüdükçe CPU kapasitesi, bellek ve depolama aktarım hızı sınırlamaları nedeniyle bunları tek bir makinede verimli biçimde işlemek zorlaşabilir.

Apache Spark, iş yüklerini bölümlere ayırabilen ve görevleri CPU çekirdekleri ile birden fazla makine arasında paralel olarak yürütebilen dağıtık bir veri işleme altyapısı sunarak bu zorlukların üstesinden gelir.

Ancak Spark, yalnızca daha fazla makine eklemekten ibaret değildir.

Bileşenlerinin (sürücüler, yürütücüler, RDD’ler, DataFrame’ler, bölümler, işler, aşamalar ve görevler) birlikte nasıl çalıştığını anlamak, verimli ve güvenilir veri işlem hatları tasarlamak için çok önemlidir.

Bu makalede, uygulamalı bir e-ticaret satış toplulaştırma örneğinden yararlanarak Spark’ın mimarisini, yürütme mekanizmalarını ve performans açısından önemli noktaları ele alacağım.

1. Apache Spark neden kullanılır? — Paralel veri işleme

Apache Spark, büyük ölçekli veri işleme ve veri mühendisliği için tasarlanmış dağıtık bir hesaplama altyapısıdır.

Spark, büyük veriyi paralel veri işlemeden yararlanarak işler:

Şekil A. Apache Spark dağıtık küme mimarisine genel bakış (küme, sürücü ve çalışan düğümler); tek düğümlü yürütmeyle karşılaştırma (Kuriko IWAI tarafından oluşturulmuştur)

Tek düğümlü işlemeyle karşılaştırıldığında (sağda, Şekil A), Spark işi birden fazla düğüm kullanarak görevlere böler ve bu görevleri makinelerdeki kaynaklar arasında, kaynaklar kullanılabilir olduğunda, yürütür.

Burada dağıtık olması, birden fazla makinenin mutlaka gerekli olduğu anlamına gelmez. Spark tek bir makinede yerel kipte çalışabilir. Bu durumda bile veriler bölümlere ayrılıp paralel olarak işlenebilir.

2. Apache Spark’ın temel mimarisi

Spark birkaç temel bileşenden oluşur:

• Spark sürücüsü (sürücü): Spark uygulamasının merkezi koordinasyon süreci.
• Küme yöneticisi: CPU çekirdekleri ve bellek gibi donanım kaynaklarını tahsis etmekten sorumlu harici hizmet.
• Çalışan düğümler (çalışan): Kümedeki fiziksel veya sanal bir makine ya da hesaplama düğümü.
• Yürütücüler: Çalışan düğümlerde başlatılan dağıtık çalışan süreçler.
• Java Virtual Machine (JVM): Spark’ın bayt kodunu yürütmek için gereken çalışma zamanı ortamı (Spark, Scala/Java ile yazıldığından).

Önce sürücü, bir Spark uygulamasını koordine eder ve onu fiziksel yürütme görevlerinden oluşan bir yönlü çevrimsiz grafa (DAG) dönüştürür; ardından görevleri yürütücüler arasında planlar.

Ardından küme yöneticisi, dağıtımın kaynak yönetimi sistemi kapsamında kaynakları tahsis eder.

Çalışan düğümler, yürütücülere hesaplama kaynakları sağlar. Yürütücüler görevleri yürütür ve görevlerin yürütülmesi sırasında verileri tutar.

Databricks gibi bazı yönetilen ortamlarda her çalışan için bir yürütücü kullanılır; ancak bu, Spark için geçerli olan evrensel bir kural değildir. Küme yapılandırmasına bağlı olarak bir çalışan birden fazla yürütücü barındırabilir.

2-1. E-ticaret kullanım örneği — Girdi verilerinin yapısı

Butik bir e-ticaret şirketi olduğunu varsayalım.

Her birinde bir stok tutma birimi (SKU) ve satılan miktar bilgisi bulunan 10.000 satış kaydı olduğunu düşünelim. Olası üç SKU bulunduğunu varsayalım: A, B, ve C.

Tablo 1. E-ticaret satış kayıtlarının dağılımı ve beklenen toplulaştırma örneği.

Amacımız, her SKU için satılan geçerli birimlerin toplamını hesaplamak:

1. Şu … olan kayıtları filtreleyin qty <= 0, ardından
2. Örnekleri SKU’ya göre toplulaştırın.

Örnek niteliğindeki tam veri kümesinde, filtreleme sonrasında 10.000 kaydın 9.800’ünün kaldığını varsayalım. Spark’tan beklenen nihai toplamlar şunlardır:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Spark kümesi

Kullanım örneğimiz için basitleştirilmiş bir küme tanımlayacağım.

Çalışan düğümler (çalışanlar)

Her biri 8 vCPU ve 32 GiB RAM’e sahip bir örnek türünde olan 3 çalışan düğüm belirleyeceğim:

Şekil B. Çalışan düğümlerde kaynak tahsisi (Kuriko IWAI tarafından oluşturulmuştur)

Sistemde toplam 24 vCPU kullanılabilir.

Spark yürütücülerine CPU/bellek tahsisi

Şöyle bir tahsis yapacağımı varsayalım:

• 7 yürütücü.
• Her yürütücü için 2 vCPU çekirdeği.
• Her yürütücü için 3 GiB JVM yığın belleği.

Spark uygulaması, küme yöneticisinden bu 7 yürütücüye kaynak tahsis etmesini ister.

Ardından küme yöneticisi, her worker’ın kaynak kullanımına göre CPU ve bellek tahsislerine karar verir.

Örneğin, 14 görev yuvası oluşturmaya karar verir: her yürütücü için 2 vCPU çekirdeği x 7 yürütücü:

Şekil C-1. Worker A’da kaynak dengesizliği varken worker’lar genelinde vCPU çekirdekleri, bellek ve yürütücü tahsisi

Yürütücüye tahsis edilen 3 GiB belleğin toplam belleği olmadığını unutmayın; çünkü worker’lar ve diğer yürütmeler için ek bellek ayırması gerekir.

Tek bir görevin 1 vCPU gerektirdiğini varsayarsak (spark.task.cpus = 1) her worker kendi vCPU’larını şöyle tahsis eder:

Şekil C-2. Worker’lar genelinde paralel vCPU tahsisi

Bu, iki önemli noktayı gösteriyor:

1. Her worker’ın kendine özgü CPU kullanım oranı vardır (Şekil C-2‘de Worker A’nın kullanımı %75 iken Worker B ve C’nin kullanımı %50’de kalır).
2. Sistemde 24 vCPU çekirdeği (her worker için 8 vCPU x 3 worker) olmasına rağmen yalnızca 14 vCPU çekirdeği kullanılabilir; kalan 10 çekirdek bu görevler için kullanılmadan kalır.

2-3. Spark’ın arkasındaki Java Sanal Makinesi (JVM)

Apache Spark, Java Sanal Makinesi (JVM) üzerine kuruludur.

JVM, Spark’ın bayt kodunu çalıştırmak için gereken bir çalışma ortamıdır.

Spark, Scala/Java ile yazıldığı için:

• Uygulamada Scala veya Java kullanıldığında: Kod, JVM içinde doğrudan çalışır.
• PySpark (Python) kullanıldığında: Python işlemi kodu çalıştırır, ancak JVM tabanlı bir sürücüyle Py4J gibi bir köprü üzerinden iletişim kurar. Worker düğümlerindeki Spark yürütücüleri yine bir JVM içinde çalışır; bu JVM de Python kodunu ve Python Kullanıcı Tanımlı İşlevlerini (Python UDF’leri) çalıştırmak için Python worker süreçlerini yönetir.

Her iki durumda da JVM, hem sürücü hem de yürütücüler için yürütme iş parçacıklarını, bellek tahsisini ve çöp toplamayı yönetir.

PySpark uygulamasındaki sürücü ve yürütücü JVM’leri

Bir PySpark uygulamasında Python sürücüsü, JVM’de çalışan Spark sürücüsüyle iletişim kurar;

• Python sürücüsü: Uygulama düzeyindeki PySpark kodunun yazıldığı ve çalıştırıldığı yerdir.
• Sürücü JVM’si: Spark yürütmesini ve zamanlamasını koordine eder.
• Yürütücü JVM’si: Spark görevlerini çalıştırır ve yürütme kaynaklarını yönetir. Görevler Python kodu çalıştırmayı gerektirdiğinde Python worker süreçleri devreye girer.
• İstemci JVM’si (istemci süreci veya gönderim makinesi): Yerel makinede ya da bir uç/ağ geçidi düğümünde çalışan JVM.

Şekilde gösterildiği gibi, istemci JVM’si önce sürücü JVM’sine bağlanır; bu JVM de daha sonra yürütücü JVM’leriyle koordinasyon kurar:

Şekil D. İstemci, sürücü ve yürütücü JVM’leri arasındaki genel ilişki

Buradan sonra sürücü JVM’si, uygulamayı yönetmek için Spark Sürücüsünü barındırırken yürütücü JVM’leri, bu görevleri yürütmekten sorumlu bağımsız yürütücüleri barındırır.

Her sürücü JVM’sinde, veri işleme, önbelleğe alma ve kullanıcı nesnelerini verimli biçimde yönetmek için bir JVM yığını bulunur.

2-4. Esnek Dağıtık Veri Kümelerine (RDD’ler) Derinlemesine Bakış

Esnek Dağıtık Veri Kümesi (RDD), Spark’ın paralel olarak işleyebildiği değiştirilemez, bölümlere ayrılmış bir öğe koleksiyonudur.

RDD’ler, dönüşümler arasındaki bağımlılıkları izleyerek Spark’ın birçok hata senaryosunda kaybolan bölümleri yeniden hesaplamasını sağlar.

Adı, üç özelliği yansıtır:

• Esnek: Spark, köken bilgisini (hangi RDD’nin hangi önceki RDD’ye bağlı olduğunu) kaydeder ve çoğu zaman her ara sonucun kopyasını tutmak yerine eksik bölümleri yeniden oluşturabilir.
• Dağıtık: Veri kümesi, Spark’ın farklı yürütücülerde işleyebileceği bölümler hâlinde temsil edilir. Birden fazla fiziksel makine zorunlu değildir; yerel mod da RDD’leri destekler.
• Veri kümesi: Bir RDD; yalnızca ilişkisel satırları değil, anahtar-değer çiftlerini, tam sayıları, dizeleri, demetleri ve desteklenen diğer öğeleri içerebilir. RDD API’leri şu tür işlemleri içerir: map, filter, reduceByKey, join, ve aggregateByKey.

RDD, belirli bölümlerin kalıcı olarak belirli çalışanlara atandığını bildiren bir tanım değil, mantıksal bir koleksiyon ve hesaplama soyutlamasıdır.

Bölümler

Bölüm, verilerin mantıksal bir parçasıdır ve normalde bir aşamadaki tek bir göreve karşılık gelir.

100 bölüm oluşturduğumuzu varsayalım (başlangıçta 100 bölüm oluşturmak, öğretim amacıyla açıkça yapılmış bir tercihtir).

10.000 kaynak öğesi varsa, girdi kabaca eşit dağıtıldığı takdirde bölüm başına yaklaşık 100 öğe düşer:

Şekil E. Başlangıçtaki 100 veri bölümünün paralel olarak dağıtılması

3. RDD API’si ve DataFrame API’si: Catalyst ile dahili optimizasyon

İlk RDD örneği, soy zincirini, bölümleri ve karıştırmayı anlamaya yardımcı olur.

Ancak tipik yapılandırılmış ETL süreçlerinde başlangıç noktası olarak DataFrame API’si tercih edilir; çünkü sütunlar, şema bilgileri, SQL benzeri ifadeler ve iyileştiricinin görebildiği işlemler sunar.

Aşağıdaki şema, aralarındaki ilişkiyi gösteriyor:

Şekil F. Spark’ın grafik kavramlarının birbiriyle ilişkisi: RDD soy zinciri, DataFrame ve Spark yürütme DAG’si

RDD’nin dönüşümleri tanımlamak için bir RDD soy zinciri olması gibi, DataFrame sorgu planı da yapılandırılmış sorgusunu temsil eder.

Spark, bağımlılıklardan ve fiziksel işleçlerden yararlanarak yürütme işlerini oluşturur.

DataFrame yürütmesi, dahili olarak RDD tabanlı mekanizmalardan yararlanabilir; ancak bu, DataFrame kodu yazarken kullanıcıların RDD soy zincirini elle oluşturması gerektiği anlamına gelmez.

3-1. DataFrame sorgu planı

DataFrame sorgu planı, yürütme motorunun oluşturduğu mantıksal ve fiziksel adımlar dizisidir (Şekil L‘deki mor kutu).

DataFrame sorgusu birkaç farklı gösterimden geçer:

1. Mantıksal plan: İstenen işlemlerin (tarama, filtreleme, gruplama, toplama) açıklaması.
2. Çözümlenmiş ve iyileştirilmiş mantıksal plan: Spark, başvuruları çözümler ve sütun budama ya da koşul itme gibi uygulanabilir mantıksal iyileştirmeleri gerçekleştirir.
3. Fiziksel plan: Spark; taramalar, kısmi/son toplamalar, değiş tokuşlar, birleştirmeler ve sıralamalar gibi yürütülebilir işleçleri seçer.
4. Yürütme: Fiziksel plan, Spark’ın yürütme mekanizmaları kullanılarak çalıştırılır ve sonuçta yürütücülerdeki görevleri içerir.

Kullanım örneğimizde fiziksel plan şu adımları içerebilir:

• Kısmi karma toplama,
• SKU’ya göre değiş tokuş ve
• Son karma toplama.

3-2. Eylemlerin karşılaştırılması: RDD API’si ve DataFrame API’si

Örnek olarak SKU’ya göre toplamayı ele alıp RDD API’siyle DataFrame API’sini karşılaştıralım.

RDD API’si

Öğeler (sku, qty) çiftlerdir. Python toplama mantığımız açıkça tanımlanmıştır ve Spark, RDD bağımlılıklarını izler:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

DataFrame API’si

Şunu varsayalım: sales_df sütunları şunlardır: sku (dize) ve qty (sayısal):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

Burada her iki örnek de aynı iş sorununu çözer:

Geçersiz satış kayıtlarını filtrelemek ve geçerli miktarları SKU’ya göre toplamak.

Her iki durumda da collect() Spark yürütmesini başlatır ve elde edilen tüm kayıtları Driver’a döndürür.

Buradaki temel fark, sonuçların gösterim biçimidir:

• RDD API’si: Şu tür Python öğeleri döndürür: (sku, total_qty) demetler.
• DataFrame API’si: Yapılandırılmış Row adlandırılmış alanları olan nesneler döndürür; örneğin sku ve total_qty.

RDD ve DataFrame API’lerinin genel karşılaştırması şöyle:

Tablo 2. RDD API ile DataFrame API – özellik ve soyutlama karşılaştırması: (Hazırlayan: Kuriko IWAI)

4. Sonuç

Spark ayarlarını iyileştirmenin amacı, bölüm sayısını veya CPU kullanımını tek başına en üst düzeye çıkarmak değildir.

Amaç, kararsızlığa yol açmadan uçtan uca yürütme maliyetini ve gecikmeyi azaltmaktır.

Bu makalede, Spark sistemindeki temel ilişkileri öğreniyoruz:

• Kaynaklar: Çalışan düğümler yürütücüleri barındırır; yürütücüler görevleri çalıştırır; yapılandırılan CPU ve bellek, eşzamanlılık sınırını belirler.
• RDD programlama: Dönüşümler, değiştirilemez RDD tanımları ve köken bilgisi üretir; eylemler hesaplamayı tetikler.
• DataFrame programlama: Dönüşümler, mantıksal sorgu işlemleri üretir; iyileştirici, yürütülebilir fiziksel işleçleri seçer; eylemler hesaplamayı tetikler.
• Yürütme: Spark, işleri aşamalara ve görevlere ayırır; görevler de yürütücülerde bölümleri işler. Shuffle işlemleri ve diğer bağımlılıklar, aşamalar arasında eşgüdüm sınırları oluşturur.

Spark’ın dağıtık iş yüklerini nasıl yürüttüğünü anlamak yalnızca temeldir.

RDD API kullanan yedi adımlı bir hesaplama akışını öğrenebilirsiniz.