PySpark ETL Integration Testing

Entegrasyon Testi Nedir ve Neden Gereklidir?

Önceki derslerde yazdığımız testler, bir motorun tek bir pistonunun (tek bir dönüşüm fonksiyonu) düzgün çalıştığını doğrulamak gibiydi. Entegrasyon testi ise, tüm pistonların, şanzımanın ve tekerleklerin birlikte uyum içinde çalışarak arabayı ileri götürüp götürmediğini (tüm ETL adımlarının birlikte çalışıp doğru çıktıyı üretip üretmediğini) kontrol etmektir.

Konuyu daha iyi anlamak için aşağıdaki iki kısa okumayı yapmanızı rica ediyorum.

Bu kaynaklardan da gördüğünüz gibi entegrasyon testleri:

• Farklı dönüşüm adımlarının birbiriyle uyumlu çalışıp çalışmadığını doğrular.
• Bir bütün olarak ETL işinin iş mantığını (business logic) doğru bir şekilde uyguladığını garanti eder.
• Girdi verisinden nihai çıktıya kadar tüm akışı test ederek daha fazla güven sağlar.

Bir entegrasyon testi genellikle üç temel adımdan oluşur: Given-When-Then (veya Arrange-Act-Assert).

1. Given (Arrange): Testin çalışması için gerekli ortamı ve girdileri hazırla. (Örn: Örnek input DataFrame'leri oluştur).
2. When (Act): Test edilecek olan asıl kodu (ETL işini) çalıştır.
3. Then (Assert): Çıktının beklenen sonuçlarla eşleşip eşleşmediğini doğrula.

Test Edeceğimiz ETL İşi

Entegrasyon testimizi yazmadan önce, test edecek bir ETL işine ihtiyacımız var. Projemizin src klasörü altında jobs adında yeni bir klasör ve içine user_reporting.py adında bir dosya oluşturalım.

pyspark-testing-project/
├── src/
│   ├── transformations.py
│   └── jobs/
│       └── user_reporting.py  <-- Yeni ETL işimiz
└── tests/
    ├── conftest.py
    ├── quality_checks.py
    ├── test_transformations.py
    └── test_data_quality.py

user_reporting.py dosyasının içeriği aşağıdaki gibi olsun. Bu iş, kullanıcıları ve harcamalarını alıp, toplam harcamalarını hesaplayarak basit bir rapor oluşturuyor.

# src/jobs/user_reporting.py

from pyspark.sql import DataFrame
from pyspark.sql.functions import sum as _sum, col

def generate_user_report(users_df: DataFrame, purchases_df: DataFrame) -> DataFrame:
    """
    Kullanıcılar ve harcamalarını birleştirip her kullanıcı için toplam harcamayı hesaplar.
    """



    # 1. Adım: Kullanıcılar ve harcamaları 'user_id' üzerinden birleştir
    enriched_purchases = purchases_df.join(users_df, "user_id", "inner")




    # 2. Adım: Kullanıcıya göre grupla ve harcamaları topla
    user_spending = enriched_purchases.groupBy("user_id", "full_name") \
                                      .agg(_sum("price").alias("total_spent"))




    # 3. Adım: Sadece belirli bir miktar üzerinde harcama yapanları filtrele
    final_report = user_spending.filter(col("total_spent") > 100.0)

    return final_report

Entegrasyon Testini Yazmak

Şimdi bu ETL işini test etmek için tests klasöründe test_user_reporting_job.py adında yeni bir test dosyası oluşturalım.

Yukarıdaki videodaki yapıyı takip ederek testimizi yazalım.

# tests/test_user_reporting_job.py

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType




# Test edilecek ETL işini import et
from src.jobs.user_reporting import generate_user_report




# Önceki derste yazdığımız kalite kontrol fonksiyonlarını import et
from tests.quality_checks import assert_row_count, assert_schema_matches

def test_generate_user_report_integration(spark_session: SparkSession):
    """
    'generate_user_report' ETL işi için entegrasyon testi.
    """



    # 1. GIVEN (Arrange): Gerekli girdileri ve beklenen çıktıyı hazırla




    # Girdi 1: users_df
    users_schema = StructType([
        StructField("user_id", StringType(), False),
        StructField("full_name", StringType(), True)
    ])
    users_data = [
        ("u1", "Alice Smith"),
        ("u2", "Bob Johnson"),
        ("u3", "Charlie Brown") # Bu kullanıcı hiç harcama yapmadı
    ]
    users_df = spark_session.createDataFrame(users_data, users_schema)




    # Girdi 2: purchases_df
    purchases_schema = StructType([
        StructField("purchase_id", StringType(), False),
        StructField("user_id", StringType(), False),
        StructField("price", DoubleType(), True)
    ])
    purchases_data = [
        ("p1", "u1", 120.50), # Alice'in harcaması > 100
        ("p2", "u2", 50.0),   # Bob'un toplam harcaması < 100
        ("p3", "u2", 30.0),
        ("p4", "u1", 80.0)    # Alice'in ikinci harcaması
    ]
    purchases_df = spark_session.createDataFrame(purchases_data, purchases_schema)




    # Beklenen Çıktı: expected_df
    expected_schema = StructType([
        StructField("user_id", StringType(), False),
        StructField("full_name", StringType(), True),
        StructField("total_spent", DoubleType(), True)
    ])
    expected_data = [
        ("u1", "Alice Smith", 200.50) # 120.50 + 80.0



        # Bob'un toplamı 80.0 olduğu için filtrelenmeli
        # Charlie hiç harcama yapmadığı için join'de kaybolmalı
    ]
    expected_df = spark_session.createDataFrame(expected_data, expected_schema)




    # 2. WHEN (Act): ETL işini çalıştır
    actual_df = generate_user_report(users_df, purchases_df)




    # 3. THEN (Assert): Çıktıyı doğrula
    



    # a) Önceki dersten gelen kalite kontrolleri ile şema ve satır sayısını kontrol et
    assert_schema_matches(actual_df, expected_schema)
    assert_row_count(actual_df, 1)




    # b) Veri içeriğinin birebir aynı olduğunu kontrol et
    # Küçük test verileri için .collect() kullanmak kabul edilebilir bir yöntemdir.
    # .collect() sonucu satır sıralaması garanti edilmediği için, karşılaştırmadan önce sıralamak iyi bir pratiktir.
    



    # Sıralamayı garanti altına al
    actual_data = sorted([row.asDict() for row in actual_df.collect()], key=lambda x: x['user_id'])
    expected_data_sorted = sorted([row.asDict() for row in expected_df.collect()], key=lambda x: x['user_id'])

    assert actual_data == expected_data_sorted, \
        f"Veri içeriği eşleşmiyor! Beklenen: {expected_data_sorted}, Gelen: {actual_data}"

Testi çalıştırmak için terminalde projenizin ana dizinindeyken pytest komutunu verin. Testin başarıyla geçtiğini göreceksiniz. Bu test, ETL işimizin birden fazla adımının (join, groupBy, agg, filter) bir arada doğru çalıştığını kanıtlar.

Tuzağa Dikkat: "Her Şeyi Test Eden" Tek Bir Büyük Test

Entegrasyon testleri güçlüdür, ancak bir tuzağa düşme riski vardır: tüm olası senaryoları tek bir büyük, karmaşık testte birleştirmeye çalışmak. Bu tür testler kırılgandır, bakımı zordur ve bir hata olduğunda sorunun kaynağını bulmayı zorlaştırır.

Bu konuyla ilgili gerçek bir hayat hikayesini dinleyelim.

Bu videodan alınacak ders şudur: Tek bir ETL işi için birden fazla, daha küçük ve odaklı entegrasyon testi yazmak genellikle daha iyidir. Örneğin:

• Bir test, harcama limiti üzerindeki kullanıcıları doğrular (bizim yazdığımız gibi).
• Ayrı bir test, harcama limiti altındaki kullanıcıların çıktıda olmadığını doğrular.
• Başka bir test, hiç harcama yapmayan kullanıcıların çıktıda olmadığını doğrular.
• Bir diğeri, girdi DataFrame'lerinden biri boş olduğunda işin hata vermeden boş bir rapor üretip üretmediğini kontrol eder.

Bu yaklaşım, testlerinizi daha okunabilir, bakımı daha kolay ve hataları daha hızlı tespit edilebilir hale getirir.

Sonuç

Bugün, bir PySpark ETL işinin tamamını test ederek test stratejimizde önemli bir adım attık. Entegrasyon testlerinin, kodumuzun farklı parçalarının bir bütün olarak uyum içinde çalıştığına dair bize nasıl güvence verdiğini gördük.

Temel Çıkarımlar:

• Entegrasyon testleri, birim testlerinin aksine, birden çok bileşenin bir arada çalışmasını doğrular ve ETL işlerinin bütünsel doğruluğu için kritiktir.
• Given-When-Then (Arrange-Act-Assert) yapısı, entegrasyon testlerini structure etmek için standart ve etkili bir yöntemdir.
• Önceki derste yazdığımız veri kalitesi doğrulama fonksiyonları, entegrasyon testlerinin "Then" (Assert) adımında çıktının kalitesini garanti etmek için doğrudan kullanılabilir.
• Tek bir devasa test yerine, her biri farklı bir iş senaryosuna veya kuralına odaklanan birden çok küçük entegrasyon testi yazmak, daha sağlam ve sürdürülebilir bir test süiti oluşturur.

Gelecek Dersin Önizlemesi:
Bugünkü testimiz, tüm verileri test fonksiyonunun içinde oluşturduğumuz için tamamen kendine yeterliydi. Peki ya ETL işimiz bir veritabanından veri okuyor veya S3'e dosya yazıyorsa? CI işlem hattımızın gerçek veritabanlarına veya bulut depolama alanlarına bağlanmasını istemeyiz. Bir sonraki dersimizde, bu tür harici bağımlılıkları "mock" etmeyi, yani sahte versiyonlarını oluşturarak testimizi izole ve hızlı tutmayı öğreneceğiz.