Tests unitaires d'un pipeline de données

ETL et ELT en Python

Jake Roach

Data Engineer

Valider un pipeline de données avec des tests unitaires

Tests unitaires :

Courants dans les flux d'ingénierie logicielle
Vérifient que le code fonctionne comme prévu
Aident à valider les données

Cadre type de validation d'un pipeline de données, incluant des tests unitaires et de bout en bout.

pytest pour les tests unitaires

from pipeline import extract, transform, load

# Créer un test unitaire qui vérifie le type de clean_stock_data
def test_transformed_data():
    raw_stock_data = extract("raw_stock_data.csv")
    clean_stock_data = transform(raw_data)
    assert isinstance(clean_stock_data, pd.DataFrame)

> python -m pytest

test_transformed_data .                                                     [100%]
================================ 1 passed in 1.17s ===============================

assert et isinstance

pipeline_type = "ETL"

# Vérifier si pipeline_type est une instance de str
isinstance(pipeline_type, str)

True

# Vérifier que la valeur du pipeline est bien "ETL"
assert pipeline_type == "ETL"

# Combiner assert et isinstance
assert isinstance(pipeline_type, str)

AssertionError

pipeline_type = "ETL"

# Provoquer un AssertionError
assert isinstance(pipeline_type, float)

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
AssertionError

Simuler des composantes de pipeline avec des fixtures

import pytest

@pytest.fixture()
def clean_data():
    raw_stock_data = extract("raw_stock_data.csv")
    clean_stock_data = transform(raw_data)
    return clean_stock_data

def test_transformed_data(clean_data):
    assert isinstance(clean_data, pd.DataFrame)

Tester des DataFrames avec des tests unitaires

def test_transformed_data(clean_data):
    # Ajouter d'autres assertions ici
    ...

    # Vérifier le nombre de colonnes
    assert len(clean_data.columns) == 4

    # Vérifier la borne inférieure d'une colonne
    assert clean_data["open"].min() >= 0

    # Vérifier l'étendue d'une colonne en enchaînant avec "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

Passons à la pratique !

ETL et ELT en Python