Implementowanie strategii testowania

Testowanie stanowi kluczową podstawę niezawodnych rozwiązań inżynierii danych. Podczas implementowania kompleksowej strategii testowania można wcześnie identyfikować problemy, weryfikować założenia i zapewnić, że potoki dostarczają wyników, które są spójne i wiarygodne. W miarę zwiększania się ilości danych, a potoki stają się coraz bardziej złożone, zautomatyzowane testowanie staje się niezbędne do utrzymania jakości i zmniejszenia ryzyka awarii produkcyjnych.

W tej lekcji dowiesz się, jak zaimplementować strategię testowania obejmującą testy jednostkowe, testy integracji, kompleksowe testy i testy akceptacyjne użytkowników (UAT) w usłudze Azure Databricks.

Omówienie piramidy testowej

Dobrze zaprojektowana strategia testowania jest zgodna z koncepcją piramidy testowej . W bazie masz wiele szybkich, izolowanych testów jednostkowych. W górę masz mniej testów integracji, które weryfikują interakcje składników. U góry znajduje się niewielka liczba kompleksowych testów end-to-end oraz scenariuszy UAT.

Ta struktura istnieje, ponieważ różne typy testów służą różnym celom:

Począwszy od testów jednostkowych, możesz mieć pewność co do podstawowych bloków konstrukcyjnych kodu. Testy integracji następnie potwierdzają, że te bloki łączą się prawidłowo. Testy end-to-end weryfikują poprawne działanie całego systemu dostarczając właściwe wyniki. Funkcja UAT zapewnia uczestnikom projektu zatwierdzenie rozwiązania przed wdrożeniem produkcyjnym.

Implementowanie testów jednostkowych za pomocą narzędzia pytest

Testy jednostkowe koncentrują się na testowaniu poszczególnych funkcji w izolacji. W Azure Databricks struktura pytest umożliwia pisanie i uruchamianie testów jednostkowych dla kodu Python.

Rozważ funkcję przekształcania danych, która filtruje rekordy według kraju/regionu:

def filter_country_region(df, country_region="USA"):
    return df[df.iso_code == country_region]

Aby przetestować tę funkcję, utwórz plik testowy zgodny z konwencjami nazewnictwa pytest. Pliki powinny zaczynać się od test_ lub kończyć się _test.py.

import pytest
import pandas as pd
from transforms import filter_country_region

@pytest.fixture
def sample_data():
    """Create test data that mimics production structure."""
    return pd.DataFrame({
        'iso_code': ['USA', 'USA', 'CAN', 'GBR'],
        'value': [100, 200, 150, 175]
    })

def test_filter_country_region_default(sample_data):
    result = filter_country_region(sample_data)
    assert len(result) == 2
    assert all(result.iso_code == 'USA')

def test_filter_country_region_specific(sample_data):
    result = filter_country_region(sample_data, country_region='CAN')
    assert len(result) == 1
    assert result.iloc[0]['value'] == 150

Dekorator @pytest.fixture tworzy dane testowe wielokrotnego użytku. Takie podejście chroni dane produkcyjne przy użyciu syntetycznych zestawów danych , które odzwierciedlają rzeczywistą strukturę danych bez ujawniania poufnych informacji.

Aby uruchomić te testy w notesie usługi Databricks, zainstaluj narzędzie pytest i wykonaj je:

%pip install pytest

import pytest
retcode = pytest.main([".", "-v", "-p", "no:cacheprovider"])
assert retcode == 0, "Tests failed. Check the output above."

Projektowanie testów integracji

Testy integracji sprawdzają, czy wiele składników działa prawidłowo. W potokach danych te testy potwierdzają, że dane przepływa prawidłowo między etapami pozyskiwania, przekształcania i magazynowania.

W przeciwieństwie do testów jednostkowych, które używają pozorowanych danych, testy integracji często są uruchamiane względem rzeczywistych zasobów usługi Databricks. Możesz odczytać z tabeli testowej, zastosować przekształcenia i zweryfikować format danych wyjściowych:

def test_pipeline_integration(spark):
    """Test that transformation pipeline produces expected schema."""
    # Read from test table
    input_df = spark.sql("SELECT * FROM test_catalog.test_schema.raw_data")

    # Apply transformation pipeline
    result_df = transform_pipeline(input_df)

    # Verify output structure
    expected_columns = ['id', 'processed_date', 'category', 'amount']
    assert result_df.columns == expected_columns

    # Verify data types
    assert result_df.schema['amount'].dataType.simpleString() == 'decimal(10,2)'

Testy integracji wymagają więcej konfiguracji niż testy jednostkowe. Utwórz dedykowane schematy testowe lub wykazy zawierające reprezentatywne przykłady danych. Ta izolacja uniemożliwia testowanie wpływu na dane produkcyjne, a jednocześnie weryfikowanie rzeczywistych interakcji składników.

Tworzenie testów end-to-end

Kompleksowe testy symulują kompletne przepływy pracy od początku do końca. Te testy sprawdzają, czy cały potok generuje oczekiwane wyniki w przypadku określonych danych wejściowych.

Kompleksowe testy struktury obejmujące realistyczne scenariusze:

def test_daily_processing_pipeline():
    """Validate complete daily data processing workflow."""
    # Setup: Create test input files
    test_date = "2024-01-15"
    setup_test_input_files(test_date)

    # Execute: Run the complete pipeline
    run_daily_pipeline(test_date)

    # Verify: Check final output table
    result = spark.sql(f"""
        SELECT COUNT(*) as row_count, 
               SUM(amount) as total_amount
        FROM test_catalog.test_schema.daily_summary
        WHERE process_date = '{test_date}'
    """)

    row = result.first()
    assert row.row_count == 1000, f"Expected 1000 rows, got {row.row_count}"
    assert abs(row.total_amount - 50000.00) < 0.01

    # Cleanup: Remove test data
    cleanup_test_data(test_date)

Kompleksowe testy mogą trwać dłużej niż testy jednostkowe lub integracyjne. Zaplanuj ich uruchamianie poza godzinami szczytu lub w ramach potoku wdrażania , a nie w przypadku każdej zmiany kodu.

Planowanie testowania akceptacyjnego użytkowników

Testowanie akceptacji użytkowników (UAT) obejmuje uczestników projektu sprawdzającego , czy rozwiązanie spełnia wymagania biznesowe. Podczas gdy poprzednie typy testów koncentrują się na poprawności technicznej, funkcja UAT potwierdza, że rozwiązanie zapewnia wartość biznesową.

Efektywna funkcja UAT wymaga starannego planowania:

• Definiowanie kryteriów akceptacji z uczestnikami projektu przed rozpoczęciem opracowywania
• Utwórz środowisko przejściowe , które odzwierciedla środowisko produkcyjne
• Przygotowywanie scenariuszy testowych , które odzwierciedlają rzeczywiste przypadki użycia biznesowego
• Dokumentowanie oczekiwanych wyników dla każdego scenariusza
• Ustanawianie procesu przesyłania opinii na potrzeby raportowania problemów

Funkcja UAT często działa w środowisku przejściowym , w którym użytkownicy biznesowi mogą korzystać z rozwiązania przy użyciu realistycznych danych. Rozważ utworzenie notesów, które uczestnicy projektu mogą wykonać w celu zweryfikowania określonych scenariuszy:

# UAT Scenario: Monthly revenue reconciliation
# Expected outcome: Total matches finance system within 1%

finance_total = 1_250_000.00  # From finance system

pipeline_result = spark.sql("""
    SELECT SUM(revenue) as total_revenue
    FROM reporting.monthly_revenue
    WHERE month = '2024-01'
""").first()

variance = abs(pipeline_result.total_revenue - finance_total) / finance_total
print(f"Variance: {variance:.2%}")

if variance < 0.01:
    print("✓ UAT PASSED: Revenue totals match within tolerance")
else:
    print("✗ UAT FAILED: Revenue variance exceeds 1% threshold")

Organizowanie struktury testów

Dobrze zorganizowana struktura testowa ułatwia konserwację i uruchamianie testów. Postępuj zgodnie z tymi konwencjami dla projektów usługi Azure Databricks:

project/
├── src/
│   └── transforms.py
├── tests/
│   ├── unit/
│   │   └── test_transforms.py
│   ├── integration/
│   │   └── test_pipeline.py
│   └── e2e/
│       └── test_daily_workflow.py
├── notebooks/
│   └── run_tests.py
└── requirements.txt

Usługa Databricks zaleca przechowywanie funkcji i testów jednostkowych poza notesami dla projektów języka Python. Takie podejście umożliwia lepsze ponowne użycie kodu i sprawia, że testowanie jest prostsze. Zapisz funkcje w plikach w .py folderze Git i zaimportuj je do notesów zgodnie z potrzebami.

W przypadku uruchamiania testów utwórz dedykowany notes, który wykonuje narzędzie pytest:

%pip install -r ../requirements.txt

import pytest
import sys

# Prevent pytest from caching to readonly filesystem
sys.dont_write_bytecode = True

# Run all tests with verbose output
retcode = pytest.main([
    "tests/",
    "-v",
    "-p", "no:cacheprovider"
])

assert retcode == 0, "Test suite failed"

Zautomatyzuj wykonywanie testów , tworząc zadanie usługi Databricks uruchamiające notes testowy przed wdrożeniem zmian w środowisku produkcyjnym. Dzięki tej automatyzacji zmiany kodu nie wprowadzają regresji.