Reusable SparkSession with Pytest Fixtures

Testlerin Ön Koşullarını Yönetmek: pytest Fixture'ları

Önceki derste kullandığımız "Arrange-Act-Assert" modelini hatırlayalım. "Arrange" adımı, test için gerekli girdileri ve ortamı hazırlamaktı. Bazen bu hazırlık adımı, basit bir değişken atamaktan daha karmaşık olabilir ve birden fazla testte tekrarlanması gerekebilir. Bir veritabanı bağlantısı kurmak, bir API istemcisi oluşturmak veya bizim durumumuzda bir SparkSession başlatmak gibi.

pytest fixture'ları, bu tür hazırlık ve temizlik (setup/teardown) işlemlerini yönetmek için tasarlanmış güçlü bir mekanizmadır. Bir fixture, @pytest.fixture dekoratörü ile işaretlenmiş bir fonksiyondur. Bu fonksiyon, bir testin ihtiyaç duyduğu bir kaynağı (veri, nesne, servis vb.) hazırlar ve teste sunar.

Fixture'ların temel faydaları şunlardır:

• Yeniden Kullanılabilirlik: Aynı hazırlık kodunu tekrar tekrar yazmak yerine, bir fixture tanımlayıp birçok testte kullanabilirsin.
• Ayrıştırma (Decoupling): Test mantığını, test ortamının kurulum mantığından ayırır. Testleriniz daha temiz ve odaklanmış olur.
• Paylaşım ve Kapsam (Scope): Bir fixture'ın ne sıklıkla çalışacağını kontrol edebilirsin. Örneğin, her test fonksiyonu için mi, her test sınıfı için mi, yoksa tüm test oturumu (session) boyunca sadece bir kez mi çalışacağı gibi.

Ortak Fixture'lar için conftest.py

Projeniz büyüdükçe, birçok test dosyasında aynı fixture'lara ihtiyaç duyacaksınız. pytest, bu ortak fixture'ları conftest.py adında özel bir dosyada merkezi olarak yönetmenize olanak tanır. tests klasörünüzün köküne yerleştireceğiniz bir conftest.py dosyasına eklediğiniz fixture'lar, o klasör ve alt klasörlerdeki tüm test dosyaları tarafından otomatik olarak tanınır ve kullanılabilir. Herhangi bir import yapmanıza gerek kalmaz.

Aşağıdaki kısa okuma parçası ve görsel, conftest.py dosyasının amacını ve SparkSession gibi paylaşılan bir fixture'ı nasıl barındırdığını çok güzel özetliyor.

Yeniden Kullanılabilir SparkSession Fixture'ı Oluşturma

Şimdi, projemiz için SparkSession fixture'ını oluşturalım. Bu fixture, tüm test süreci boyunca yalnızca bir kez oluşturulacak, tüm testler tarafından kullanılacak ve en sonunda düzgün bir şekilde kapatılacaktır. Bu, testlerimizin çalışma hızını önemli ölçüde artıracaktır.

Towards Data Science üzerindeki bu makale, lokal testler için optimize edilmiş yapılandırma parametreleriyle birlikte yeniden kullanılabilir bir SparkSession fixture'ının nasıl oluşturulacağını adım adım açıklıyor.

Okuduğun makaledeki bilgiler ışığında, projemizin tests klasörünün içine conftest.py adında yeni bir dosya oluşturalım ve aşağıdaki kodu ekleyelim:

# tests/conftest.py

import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def spark():
    """
    Tüm test oturumu boyunca kullanılacak bir SparkSession fixture'ı.
    Lokal testler için optimize edilmiştir.
    """
    spark_session = (
        SparkSession.builder.master("local[1]")
        .appName("pyspark-unittests")
        .config("spark.executor.cores", "1")
        .config("spark.executor.instances", "1")
        .config("spark.sql.shuffle.partitions", "1")
        .config("spark.driver.bindAddress", "127.0.0.1")
        .getOrCreate()
    )
    



    # Fixture'ı testlere sunar
    yield spark_session
    



    # Test oturumu bittikten sonra SparkSession'ı durdurur
    spark_session.stop()

Bu kodda birkaç önemli nokta var:

• @pytest.fixture(scope="session"): Bu dekoratör, spark fonksiyonunu bir fixture olarak tanımlar. scope="session" parametresi, pytest'e bu fixture'ı tüm test oturumu boyunca yalnızca bir kez çalıştırmasını söyler. SparkSession gibi başlatılması maliyetli nesneler için bu ideal kapsamdır.
• Konfigürasyonlar: master("local[1]") ve spark.sql.shuffle.partitions, Spark'a tek bir iş parçacığı üzerinde çalıştığını ve dağıtık hesaplama yapmaya çalışmaması gerektiğini söyler. Bu, büyük veri kümeleri yerine küçük test verileriyle çalıştığımız yerel ortamda performansı önemli ölçüde artırır.
• yield spark_session: return yerine yield kullanmak, pytest fixture'larında önerilen bir pratiktir. yield ifadesi, spark_session nesnesini teste teslim eder. Test bittikten sonra, kod yield ifadesinden sonraki kısımdan devam eder.
• spark_session.stop(): yield'den sonraki bu satır, tüm testler tamamlandığında SparkSession'ın düzgün bir şekilde kapatılmasını sağlar. Bu, kaynak sızıntılarını önler.

Aşağıdaki video, bu konsepti canlı bir örnek üzerinde gösteriyor. Bir PySpark projesi için spark_session fixture'ının nasıl tanımlandığını ve scope='session' parametresinin neden önemli olduğunu pratik olarak görebilirsin.

Fixture'ı Testte Kullanmak

Artık spark adında oturum kapsamlı bir fixture'ımız olduğuna göre, onu testlerimizde kullanabiliriz. pytest, bunu sihirli bir şekilde kolaylaştırır: Tek yapman gereken, test fonksiyonuna fixture ile aynı adı taşıyan bir argüman eklemektir.

Şimdi tests/test_transformations.py dosyamıza, spark fixture'ımızın doğru çalışıp çalışmadığını kontrol eden basit bir test ekleyelim.

# tests/test_transformations.py




# ... önceki dersteki import ve test fonksiyonu ...
from etl_pipeline.transformations import standardize_column_name

def test_standardize_column_name():



    # ... bu test olduğu gibi kalabilir ...
    input_string = "  First Name  "
    expected_output = "first_name"
    actual_output = standardize_column_name(input_string)
    assert actual_output == expected_output




# YENİ TEST: Spark fixture'ını kullanma
def test_spark_session_is_available(spark):
    """
    'spark' fixture'ının test fonksiyonlarına doğru şekilde
    enjekte edildiğini doğrular.
    """



    # Arrange: Boş bir şema tanımla
    schema = "id INT, name STRING"
    



    # Act: Boş bir DataFrame oluştur
    df = spark.createDataFrame([], schema=schema)
    



    # Assert: DataFrame'in var olduğunu ve satır sayısının 0 olduğunu doğrula
    assert df is not None
    assert df.count() == 0

Bu test_spark_session_is_available fonksiyonuna dikkat et. spark adında bir argüman alıyor. pytest çalıştırıldığında, conftest.py dosyasındaki spark fixture'ını bulur, onu çalıştırır (SparkSession'ı başlatır) ve dönen spark_session nesnesini bu argüman yerine koyar.

Şimdi terminalde projenizin kök dizinindeyken pytest komutunu tekrar çalıştır. Çıktıda 2 passed görmelisin. Bu, hem önceki dersin basit Python testinin hem de yeni eklediğimiz ve SparkSession kullanan testin başarıyla geçtiğini gösterir.

Özet ve Sonraki Adımlar

Bu dersimizde, PySpark test altyapımızın en önemli parçalarından birini inşa ettik.

Temel Çıkarımlar:

• pytest Fixture'ları, testler için gerekli olan kaynakları (veri, nesne, servis) hazırlamak ve temizlemek için kullanılır.
• conftest.py dosyası, birden fazla test dosyası arasında paylaşılabilecek ortak fixture'ları barındırmak için ideal yerdir.
• scope="session", bir SparkSession gibi başlatılması maliyetli olan nesneleri tüm test oturumu boyunca yalnızca bir kez oluşturarak testleri hızlandırır.
• Bir fixture'ı kullanmak için, test fonksiyonuna fixture'ın adıyla eşleşen bir argüman eklemek yeterlidir.

Gelecek Dersin Önizlemesi:
Artık testlerimizde kullanabileceğimiz hazır bir SparkSession'ımız var. Bu sayede, bir sonraki dersimizde asıl amacımıza odaklanabiliriz: Tekil PySpark DataFrame dönüşüm fonksiyonları için birim testleri (unit tests) yazmak. Test verilerini içeren DataFrame'ler oluşturacak, dönüşüm fonksiyonumuzu bu verilere uygulayacak ve sonucun beklediğimiz gibi olup olmadığını assert ile kontrol edeceğiz.