Testing PySpark Jobs Locally with Docker

Adım 1: Docker İmajını Oluşturmak (Build)

Doğrulama yapmadan önce, önceki derste hazırladığımız Dockerfile'ı kullanarak bir Docker imajı oluşturmamız gerekiyor. Bu işlemi docker build komutuyla yaparız.

Şimdi, projenizin ana dizininde bir terminal açın ve aşağıdaki komutu çalıştırarak kendi PySpark ETL uygulamanız için imajı oluşturun:

docker build -t pyspark-etl:latest .

• docker build: Docker'a bir imaj oluşturma komutu verir.
• -t pyspark-etl:latest: Oluşturulan imajı etiketler. pyspark-etl imajın adıdır ve latest ise etiketidir (versiyon gibi düşünebilirsiniz).
• .: Docker'a Dockerfile'ı mevcut dizinde (.) aramasını söyler.

Bu komut tamamlandığında, docker images komutunu çalıştırarak pyspark-etl adında yeni bir imajın oluşturulduğunu görebilirsiniz.

Adım 2: Konteynerleştirilmiş İşi Çalıştırmak (Run)

İmajımız hazır olduğuna göre, şimdi onu bir konteyner olarak çalıştırabiliriz. ETL işimiz bir batch (toplu) işlem olduğu için, konteynerin başlayıp, PySpark betiğini çalıştırıp, ardından durmasını istiyoruz.

Bunu docker run komutu ile şu şekilde yaparız:

docker run <İMAJ_ADI> <ÇALIŞTIRILACAK_KOMUT>

Bizim durumumuzda, konteyner içinde spark-submit komutunu kullanarak main.py betiğimizi çalıştırmak istiyoruz. Önceki dersteki Dockerfile'ımıza göre betiğimiz /app/src/main.py yolunda olacaktır.

Dolayısıyla, en temel komut şöyledir:

docker run pyspark-etl:latest spark-submit /app/src/main.py

Veri Giriş ve Çıkışını Yönetmek: Volumes

Ancak bir sorun var: ETL işimizin işlemesi için bir girdi verisine ve sonuçları yazması için bir çıktı dizinine ihtiyacı var. Konteynerler varsayılan olarak izole dosya sistemlerine sahiptir. Yerel makinenizdeki dosyalara doğrudan erişemezler.

Bu sorunu Docker volumes (birimler) ile çözeriz. Volume, yerel makinenizdeki bir dizini konteyner içindeki bir dizine "bağlamanızı" (mount) sağlar. Bu sayede, konteyner bu dizine yazdığında, değişiklikler aslında yerel makinenizdeki dizine yansır ve tam tersi de geçerlidir.

Proje yapınızı aşağıdaki gibi düzenleyelim ve main.py betiğini buna göre güncelleyelim:

1. Proje Dizini Yapısı:
Projenizin ana dizininde data adında bir klasör oluşturun. İçine de input ve output adında iki alt klasör ekleyin. input klasörüne basit bir CSV dosyası (örneğin, iris.csv) yerleştirin.

/pyspark-ci-cd-project
|-- Dockerfile
|-- requirements.txt
|-- data/
|   |-- input/
|   |   `-- iris.csv
|   `-- output/
`-- src/
    `-- main.py

İpucu: Iris veri setini wget https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data -O data/input/iris.csv komutuyla hızlıca indirebilirsiniz.

2. main.py Güncellemesi:
src/main.py betiğinizin, veriyi volume ile bağlanacak yollardan okuyup yazdığından emin olun. Dockerfile'daki WORKDIR /app olduğu için, konteyner içindeki yollar /app ile başlamalıdır.

# src/main.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import avg

if __name__ == "__main__":
    spark = SparkSession.builder.appName("LocalETLValidation").getOrCreate()




    # Bu yollar, Docker volume'u ile eşleştirilecek
    input_path = "/app/data/input/iris.csv"
    output_path = "/app/data/output/species_avg_sepal_length"




    # Veriyi oku
    df = spark.read.csv(input_path, inferSchema=True, header=False)
    



    # Sütunları yeniden adlandır
    columns = ["sepal_length", "sepal_width", "petal_length", "petal_width", "species"]
    df = df.toDF(*columns)




    # Basit bir transformasyon yap
    avg_sepal_length = df.groupBy("species").agg(avg("sepal_length").alias("avg_sepal_length"))




    # Sonucu yaz
    avg_sepal_length.write.mode("overwrite").parquet(output_path)

    print("ETL job completed successfully. Output written to:", output_path)

    spark.stop()

3. docker run Komutunu Volume ile Çalıştırmak:
Şimdi, -v bayrağını kullanarak data klasörümüzü konteynerin içine bağlayalım.

docker run --rm \
  -v "$(pwd)/data:/app/data" \
  pyspark-etl:latest \
  spark-submit /app/src/main.py

• --rm: Konteyner durduktan sonra otomatik olarak silinmesini sağlar. Bu, test amaçlı çalıştırılan geçici konteynerler için iyi bir pratiktir.
• -v "$(pwd)/data:/app/data": Bu en önemli kısım.
  • $(pwd)/data: Yerel makinenizdeki mevcut çalışma dizininizin (pwd) altındaki data klasörünün tam yolunu belirtir.
  • :: Yerel yol ile konteyner yolunu ayırır.
  • /app/data: Bu yerel data dizininin konteyner içinde bağlanacağı hedef yoldur.

Bu komutu çalıştırdığınızda, Spark'ın loglarını ve betiğimizdeki print ifadesini görmelisiniz. İşlem bittiğinde, yerel makinenizdeki data/output dizinini kontrol edin. İçinde Parquet formatında yeni dosyaların oluştuğunu göreceksiniz. Bu, konteynerleştirilmiş işinizin başarıyla çalıştığının ve yerel dosya sisteminize veri yazabildiğinin kanıtıdır!

Adım 3: Hata Ayıklama ve Doğrulama (Debugging)

İşler her zaman yolunda gitmeyebilir. Bir hata durumunda ne yapmalıyız?

Eğer docker run komutunuz başarısız olursa, hatanın kaynağını bulmak için konteynerin içine interaktif bir shell (komut satırı) ile girebilirsiniz. Bu, size adeta konteynerin "içine ışınlanma" imkanı verir.

docker run --rm -it \
  -v "$(pwd)/data:/app/data" \
  pyspark-etl:latest \
  /bin/bash

• -it: Bu iki bayrağın birleşimidir:
  • -i (interactive): Standart girdiyi (klavyenizi) açık tutar.
  • -t (tty): Size bir terminal benzeri arayüz sağlar.
• /bin/bash: Konteyner başladığında çalıştırılacak varsayılan komutu (spark-submit) geçersiz kılar ve yerine bir bash shell'i başlatır.

Bu komutu çalıştırdıktan sonra, terminal prompt'unuzun değiştiğini ve artık konteynerin içinde olduğunuzu göreceksiniz. Burada şunları yapabilirsiniz:

• ls -l /app/data/input: Veri dosyalarınızın doğru şekilde bağlanıp bağlanmadığını kontrol edin.
• env: Ortam değişkenlerini listeleyin (örneğin, SPARK_HOME doğru ayarlanmış mı?).
• spark-submit /app/src/main.py: Betiğinizi manuel olarak çalıştırarak daha detaylı hata mesajları almayı deneyin.

İşiniz bittiğinde exit yazarak shell'den çıkabilirsiniz.

Sonuç

Bu derste, optimize edilmiş Docker imajımızı alıp yerel olarak nasıl çalıştıracağımızı ve doğrulayacağımızı öğrendik. Bu, geliştirme döngüsünün vazgeçilmez bir parçasıdır.

Temel Çıkarımlar:

• docker build -t <ad>:<etiket> . komutu, bir Dockerfile'dan etiketlenmiş bir imaj oluşturur.
• docker run <imaj> [komut] komutu, bir imajdan yeni bir konteyner başlatır ve içinde bir komut çalıştırır.
• ETL işleri için veri giriş/çıkışını sağlamak amacıyla -v (volume) bayrağı ile yerel dizinler konteynere bağlanır.
• Bir iş başarısız olduğunda, docker run -it <imaj> /bin/bash komutu, hata ayıklama için konteynerin içine interaktif bir shell başlatmanın en etkili yoludur.

Gelecek Dersin Önizlemesi:
Artık yerelde çalışan, doğrulanmış bir konteynerimiz var. Bir sonraki adım, bu süreci otomatikleştirmektir. Gelecek derste, kodumuzu her push ettiğimizde bu Docker imajını otomatik olarak oluşturacak ve bir container registry'e (bu durumda GitHub Container Registry) gönderecek bir GitHub Actions iş akışı (workflow) adımı ekleyeceğiz. Bu, CI/CD işlem hattımızın "build and push" aşaması olacak.