Unit Testing PySpark DataFrame Transformations

PySpark Birim Testinin Anatomisi: Arrange, Act, Assert

Bir PySpark dönüşüm fonksiyonunu test etme süreci, standart "Arrange-Act-Assert" modelini takip eder, ancak DataFrame'ler ile çalışmanın getirdiği bazı özellikler vardır:

1. Arrange (Hazırla):

   • Test etmek istediğiniz dönüşüm fonksiyonu için küçük, kontrollü bir girdi (input) DataFrame'i oluşturun.
   • Dönüşüm uygulandıktan sonra elde etmeyi beklediğiniz sonucu içeren bir beklenen (expected) DataFrame oluşturun.
   • Bu adımların her ikisi için de önceki derste oluşturduğumuz spark fixture'ını kullanacağız.

2. Act (Uygula):

   • Test edilecek dönüşüm fonksiyonunu, hazırladığınız girdi DataFrame'i ile çağırın. Bu çağrının sonucunda bir gerçekleşen (actual) DataFrame elde edeceksiniz.

3. Assert (Doğrula):

   • Gerçekleşen DataFrame'in, beklenen DataFrame ile aynı olup olmadığını kontrol edin. Bu, PySpark testlerinin en kritik kısmıdır. Sadece satır sayısını kontrol etmek yeterli değildir; şema (schema), veri ve hatta bazen satır sırası gibi unsurları da doğrulamamız gerekir.

PySpark için Pratik Bir Test Senaryosu

Teoriyi pratiğe dökelim. Projemizin etl_pipeline/transformations.py dosyasında, müşteri verilerini zenginleştiren fonksiyonlarımız olduğunu varsayalım. İşte test edeceğimiz basit bir dönüşüm fonksiyonu:

Öncelikle, etl_pipeline/transformations.py dosyanıza aşağıdaki fonksiyonu ekleyin. Bu fonksiyon, first_name ve last_name sütunlarını birleştirerek full_name adında yeni bir sütun oluşturur.

# etl_pipeline/transformations.py




#... (varsa) diğer import'lar ...
from pyspark.sql import DataFrame
import pyspark.sql.functions as F

def standardize_column_name(column_name: str) -> str:



    # ... (önceki dersten kalan fonksiyon) ...
    return column_name.strip().lower().replace(" ", "_")

def with_full_name(df: DataFrame) -> DataFrame:
    """
    'first_name' ve 'last_name' sütunlarını birleştirerek
    'full_name' adında yeni bir sütun ekler.
    """
    return df.withColumn("full_name", F.concat_ws(" ", F.col("first_name"), F.col("last_name")))

Şimdi, bu with_full_name fonksiyonunun doğru çalışıp çalışmadığını test edelim.

DataFrame'leri Karşılaştırmak: assertDataFrameEqual

İki DataFrame'in tamamen aynı olup olmadığını manuel olarak kontrol etmek (örneğin collect() yapıp listeleri karşılaştırmak) zahmetli ve hataya açıktır. Neyse ki, PySpark'ın kendi test yardımcı programları bu işi bizim için çok kolaylaştırır. pyspark.testing.utils modülündeki assertDataFrameEqual fonksiyonu, iki DataFrame'in şemalarını ve içeriklerini karşılaştırır ve farklılık varsa ayrıntılı bir hata mesajı verir.

Aşağıdaki okuma materyali, PySpark'ın resmi dokümantasyonundan alınmıştır ve assertDataFrameEqual fonksiyonunun nasıl kullanılacağını basit bir örnekle göstermektedir.

Test Fonksiyonumuzu Yazalım

Şimdi, öğrendiklerimizi birleştirerek tests/test_transformations.py dosyamıza with_full_name fonksiyonu için birim testimizi ekleyelim.

# tests/test_transformations.py




# ... (önceki import'lar ve test fonksiyonları) ...
from etl_pipeline.transformations import with_full_name
from pyspark.testing.utils import assertDataFrameEqual




# ... test_standardize_column_name ve test_spark_session_is_available fonksiyonları burada kalabilir ...

def test_with_full_name(spark):
    """
    'with_full_name' dönüşümünün 'first_name' ve 'last_name' sütunlarını
    doğru şekilde birleştirdiğini test eder.
    """



    # 1. Arrange (Hazırla)
    input_data = [
        {"first_name": "John", "last_name": "Doe", "age": 30},
        {"first_name": "Jane", "last_name": "Smith", "age": 25},
    ]
    input_df = spark.createDataFrame(input_data)

    expected_data = [
        {"first_name": "John", "last_name": "Doe", "age": 30, "full_name": "John Doe"},
        {"first_name": "Jane", "last_name": "Smith", "age": 25, "full_name": "Jane Smith"},
    ]



    # Beklenen DataFrame'i oluştururken şemanın sırasının tutarlı olmasına dikkat edin.
    expected_df = spark.createDataFrame(expected_data, schema=input_df.schema.add("full_name", "string"))




    # 2. Act (Uygula)
    actual_df = with_full_name(input_df)




    # 3. Assert (Doğrula)
    # assertDataFrameEqual, sütun sırasına duyarlı olabilir, bu nedenle sütunları seçerek sıralarını garantileyebiliriz.
    # Ancak bu örnekte PySpark genellikle aynı sırayı korur.
    assertDataFrameEqual(actual_df, expected_df)

Şimdi terminalde projenizin kök dizinindeyken pytest komutunu çalıştırın. Tüm testlerin başarıyla geçtiğini (3 passed) görmelisiniz.

Daha Detaylı Doğrulamalar Yapmak

assertDataFrameEqual tüm DataFrame'i kontrol etmek için harikadır, ancak bazen daha spesifik doğrulamalar yapmak isteyebiliriz. Örneğin:

• Sadece belirli bir sütunun varlığını kontrol etmek.
• Belirli bir koşulu sağlayan bir satırdaki tek bir değeri doğrulamak.

Bu tür testler, mantığınızın belirli kısımlarını izole ederek daha odaklı geri bildirim sağlar.

Aşağıdaki video, PySpark dönüşümlerini test ederken bu tür spesifik doğrulamaların nasıl yapıldığını harika bir şekilde göstermektedir.

Videoda görülen tekniği kullanarak, testimizi belirli bir satırın full_name değerini kontrol edecek şekilde genişletebiliriz. Bu, assertDataFrameEqual'a ek olarak veya onun yerine kullanılabilir.

Örneğin, test_with_full_name fonksiyonunun Assert bölümünü şu şekilde de yazabilirdik:

# Örnek bir alternatif Assert bölümü




# Assert
assert "full_name" in actual_df.columns  # Yeni sütunun varlığını kontrol et




# John Doe'nun tam adını kontrol et
john_row = actual_df.filter(actual_df.first_name == "John").first()
assert john_row["full_name"] == "John Doe"

Bu yaklaşım, özellikle karmaşık dönüşümlerde, hatanın tam olarak hangi değerde veya koşulda meydana geldiğini anlamak için çok faydalıdır.

Özet ve Sonraki Adımlar

Bugün, PySpark ETL işleriniz için güvenilir ve sürdürülebilir bir test süiti oluşturmanın en temel adımını attınız.

Temel Çıkarımlar:

• Bir PySpark birim testi "Arrange-Act-Assert" modelini izler: Girdi/beklenen DataFrame'leri hazırla, dönüşümü uygula ve sonucu doğrula.
• pyspark.testing.utils.assertDataFrameEqual, iki DataFrame'in şema ve içeriğini karşılaştırmak için güçlü ve kullanışlı bir araçtır.
• Bazen tüm DataFrame'i karşılaştırmak yerine, filter(), first() ve assert kullanarak belirli sütunları veya değerleri doğrulamak daha hedefli bir yaklaşım olabilir.
• Test edilebilir kod yazmak önemlidir. PySpark mantığınızı, her biri tek bir iş yapan küçük, saf (pure) dönüşüm fonksiyonlarına bölmek, birim testlerini yazmayı çok daha kolaylaştırır.

Gelecek Dersin Önizlemesi:
Bu derste, dönüşüm fonksiyonumuzu tek bir "mutlu yol" (happy path) senaryosu ile test ettik. Peki ya verilerimiz null değerler içeriyorsa, beklenen sütunlar eksikse veya beklenmedik veri türleri varsa ne olur? Gerçek dünya verileri nadiren mükemmeldir.

Bir sonraki dersimizde, pytest'in parametrize özelliğini kullanarak aynı test mantığını birden çok girdi senaryosuyla (normal durumlar, sınır durumları ve hata durumları) kolayca nasıl çalıştıracağımızı öğreneceğiz. Bu, fonksiyonlarımızın daha sağlam (robust) olmasını sağlayacaktır.