PySpark Data Quality Checks

Veri Kalitesi Testlerinin Önemi

Veri işlem hatlarında hatalar yalnızca yanlış kod mantığından kaynaklanmaz. Çoğu zaman, beklenmedik veya hatalı kaynak verilerden (boş değerler, yanlış veri tipleri, eksik sütunlar) kaynaklanır. Bu tür sorunları erkenden tespit etmek, veri bozulmasını önler ve aşağı akış sistemlerinin (downstream systems) sağlığını korur.

Aşağıdaki görsel, bir veri kalitesi kontrol akışının konseptini çok güzel özetliyor. Hatalı verileri barındıran bir girdi DataFrame'i, tanımladığımız kalite kurallarından geçirilir. Kurallara uyan "iyi veriler" yoluna devam ederken, kuralları ihlal eden "kötü veriler" karantinaya alınır ve işlem hattı durdurulur. İşte biz bugün bu "kuralları" kodlayan fonksiyonları yazacağız.

Veri Kalitesi Doğrulama Fonksiyonları Tasarlamak

Amacımız, pytest testlerimizde tekrar tekrar kullanabileceğimiz, modüler doğrulama fonksiyonları oluşturmaktır. Bu yaklaşım, veri mühendisliği geçmişinizden aşina olduğunuz kodun yeniden kullanılabilirliği (reusability) ve modülerlik ilkeleriyle örtüşmektedir.

Bu fonksiyonları barındırmak için projemizin tests klasörü içinde quality_checks.py adında yeni bir dosya oluşturalım. Test mantığımızı, testlerimizin kendisinden ayırmak iyi bir pratiktir.

pyspark-testing-project/
├── src/
│   └── transformations.py
└── tests/
    ├── conftest.py
    ├── test_transformations.py
    └── quality_checks.py  <-- Yeni dosyamız

Şimdi bu dosyayı, temel kalite kontrollerini uygulayan fonksiyonlarla doldurmaya başlayalım.

1. Satır Sayısı Kontrolü (Row Count Check)

Bu en temel kontrollerden biridir. Bir dönüşümün beklenmedik şekilde satırları silmediğinden veya çoğaltmadığından emin olmak için kullanılır.

Aslında bu kontrolü daha önce bir test içinde dolaylı olarak yapmıştık. Şimdi bunu yeniden kullanılabilir bir fonksiyona dönüştürelim.

Bu mantığı kullanarak, tests/quality_checks.py dosyasına ilk fonksiyonumuzu ekleyelim:

# tests/quality_checks.py

from pyspark.sql import DataFrame

def assert_row_count(df: DataFrame, expected_count: int):
    """DataFrame'in belirtilen sayıda satıra sahip olduğunu doğrular."""
    actual_count = df.count()
    assert actual_count == expected_count, \
        f"Satır sayısı kontrolü başarısız! Beklenen: {expected_count}, Gerçekte: {actual_count}"

Bu fonksiyon bir DataFrame ve beklenen bir sayı alır, assert kullanarak kontrolü yapar. Eşitlik sağlanmazsa, pytest'in hata raporunda gösterilecek açıklayıcı bir mesajla birlikte bir AssertionError fırlatır.

2. Boş Değer Kontrolü (Null Check)

Boş (NULL veya None) değerler, veri kalitesindeki en yaygın sorunlardan biridir. Özellikle ID sütunları veya önemli metrikler gibi asla boş olmaması gereken alanlarda bu kontrol kritiktir.

Aşağıdaki makale, PySpark ile uygulanabilecek bir dizi pratik veri kalitesi kontrolü sunmaktadır.

Makaledeki fikirden ilham alarak, belirli bir sütunda hiç NULL değer olmadığını doğrulayan bir fonksiyon yazalım. tests/quality_checks.py dosyamıza ekleyelim:

# tests/quality_checks.py

from pyspark.sql import DataFrame
from pyspark.sql.functions import col, count, when




# ... (assert_row_count fonksiyonu burada)

def assert_no_nulls_in_col(df: DataFrame, column_name: str):
    """Belirtilen sütunda hiç NULL değer olmadığını doğrular."""
    null_count = df.filter(col(column_name).isNull()).count()
    assert null_count == 0, \
        f"'{column_name}' sütununda {null_count} adet NULL değer bulundu!"

Bu fonksiyon, belirtilen sütunu isNull() ile filtreler, eşleşen satırları sayar ve sayının sıfır olduğunu iddia eder.

3. Şema Doğrulaması (Schema Validation)

Veri işlem hatlarınız, bir "veri sözleşmesine" (data contract) uymalıdır. Şema doğrulaması, bir DataFrame'in beklenen sütun adlarına ve veri tiplerine sahip olduğunu garanti ederek bu sözleşmeyi uygular.

DataFrame.schema özelliği, bize bir StructType nesnesi döndürür. Bu nesneyi, beklediğimiz StructType ile doğrudan karşılaştırabiliriz. Bu, yalnızca sütun adlarını değil, aynı zamanda veri tiplerini ve nullable (boş değer alabilme) özelliklerini de kontrol eden çok sağlam bir yöntemdir.

Şimdi bu kontrolü tests/quality_checks.py içine bir fonksiyon olarak ekleyelim:

# tests/quality_checks.py

from pyspark.sql import DataFrame
from pyspark.sql.types import StructType



# ... (diğer importlar ve fonksiyonlar)

def assert_schema_matches(df: DataFrame, expected_schema: StructType):
    """DataFrame'in şemasının beklenen şema ile eşleştiğini doğrular."""



    # Sütun sırasından bağımsız karşılaştırma için şemaları normalleştirebiliriz,
    # ancak şimdilik doğrudan karşılaştırma yapacağız.
    # Doğrudan karşılaştırma hem sırayı hem de tipleri zorunlu kılar.
    assert df.schema == expected_schema, \
        f"Şema uyuşmazlığı! Beklenen: {expected_schema}, Gerçekte: {df.schema}"

Doğrulama Fonksiyonlarını Test İçinde Kullanma

Artık bir doğrulama araç setimiz var. Şimdi bunları bir pytest testi içinde nasıl kullanacağımızı görelim. tests klasöründe test_data_quality.py adında yeni bir test dosyası oluşturalım.

Bu testte, "raw_customers" adlı varsayımsal bir girdi verisinin temel kalite standartlarını karşıladığını doğrulayacağız.

# tests/test_data_quality.py

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




# Yeni doğrulama fonksiyonlarımızı import edelim
from .quality_checks import assert_row_count, assert_no_nulls_in_col, assert_schema_matches

def test_raw_customers_quality(spark_session: SparkSession):
    """
    'raw_customers' verisi için temel veri kalitesi kontrollerini çalıştırır.
    """



    # 1. Beklenen şemayı tanımla
    expected_schema = StructType([
        StructField("customer_id", StringType(), False), # False -> nullable=False
        StructField("name", StringType(), True),
        StructField("age", IntegerType(), True)
    ])




    # 2. Test verisini oluştur
    # customer_id'nin null olmaması gerekiyor (şemada nullable=False)
    # age sütununda bir null değer var, ancak bu sütun nullable=True olduğu için sorun değil.
    sample_data = [
        ("c101", "Alice", 34),
        ("c102", "Bob", None),
        ("c103", "Charlie", 45)
    ]
    
    df = spark_session.createDataFrame(sample_data, schema=expected_schema)




    # 3. Doğrulama fonksiyonlarını çalıştır
    assert_row_count(df, 3)
    assert_no_nulls_in_col(df, "customer_id")
    assert_schema_matches(df, expected_schema)




    # Kasıtlı olarak başarısız olacak bir test durumu (yorum satırını kaldırarak deneyin)
    # assert_no_nulls_in_col(df, "age")

Bu dosyayı oluşturduktan sonra, terminalde pytest komutunu çalıştırın. Yeni testinizin başarıyla geçtiğini görmelisiniz. Ardından, assert_no_nulls_in_col(df, "age") satırının yorumunu kaldırıp testi tekrar çalıştırarak, quality_checks.py içinde tanımladığınız özel hata mesajıyla testin nasıl başarısız olduğunu gözlemleyin.

Sonuç ve Sonraki Adımlar

Bu derste, veri mühendisliğindeki en önemli pratiklerden birini test suitimize entegre ettik. Artık yalnızca kodumuzun mantığını değil, aynı zamanda işlediği verinin kalitesini de programatik olarak doğrulayabiliyoruz.

Temel Çıkarımlar:

• Veri kalitesi kontrolleri (satır sayısı, null kontrolü, şema doğrulaması) veri işlem hatlarının güvenilirliği için hayati öneme sahiptir.
• Bu kontrolleri yeniden kullanılabilir fonksiyonlar halinde (quality_checks.py gibi) yazmak, test suitini modüler ve bakımı kolay hale getirir.
• assert ifadelerine eklenen özel hata mesajları, pytest raporlarının okunabilirliğini artırır ve hataların nedenini hızla anlamamızı sağlar.
• PySpark'ın DataFrame.schema ve StructType nesneleri, veri sözleşmelerini zorunlu kılmak için güçlü bir mekanizma sunar.

Kendi fonksiyonlarımızı yazmak, temel prensipleri öğrenmek için harika bir yoldur. Gerçek dünya projelerinde, bu işlevselliği ve çok daha fazlasını sunan yerleşik kütüphaneler de kullanılır. Aşağıdaki görsel, bu alandaki popüler araçlardan bazılarını karşılaştırmaktadır. Gelecekte bu tür araçları keşfetmek, platform mühendisliği hedeflerinizle de uyumlu olacaktır.

Gelecek Dersin Önizlemesi:
Bugün, statik bir DataFrame üzerinde kalite kontrolleri yaptık. Bir sonraki dersimizde, "Uçtan uca PySpark ETL iş çıktılarının beklenen sonuçlara göre doğrulandığı entegrasyon testleri yazma" konusuna geçeceğiz. Bu, bir veya daha fazla girdi DataFrame'i alıp, bir dönüşümden geçirip, ortaya çıkan çıktı DataFrame'ini doğrulayacağımız anlamına gelir. Bugün yazdığımız kalite doğrulama fonksiyonları, bu entegrasyon testlerinde çıktının kalitesini garanti altına almak için kilit bir rol oynayacak.