Tester un pipeline de données avec des tests unitaires

ETL et ELT en Python

Jake Roach

Data Engineer

Valider un pipeline de données avec des tests unitaires

Tests unitaires :

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

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

pytest pour les tests unitaires

from pipeline import extract, transform, load

# Construire un test unitaire, en vérifiant 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

# Affirmer que le pipeline a bien la valeur "ETL"
assert pipeline_type == "ETL"

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

AssertionError

pipeline_type = "ETL"

# Provoquer une AssertionError
assert isinstance(pipeline_type, float)

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

Simuler des composants 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'intervalle d'une colonne en chaînant avec "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

Passons à la pratique !

ETL et ELT en Python