Pruebas unitarias de un pipeline de datos

ETL y ELT en Python

Jake Roach

Data Engineer

Validar un pipeline de datos con pruebas unitarias

Pruebas unitarias:

Muy usadas en el desarrollo de software
Aseguran que el código funciona como se espera
Ayudan a validar datos

Marco típico de validación de pipelines de datos, con pruebas unitarias y de extremo a extremo.

pytest para pruebas unitarias

from pipeline import extract, transform, load

# Crea una prueba unitaria que verifique el tipo 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 e isinstance

pipeline_type = "ETL"

# Comprueba si pipeline_type es instancia de str
isinstance(pipeline_type, str)

True

# Verifica que el pipeline realmente vale "ETL"
assert pipeline_type == "ETL"

# Combina assert e isinstance
assert isinstance(pipeline_type, str)

AssertionError

pipeline_type = "ETL"

# Genera un AssertionError
assert isinstance(pipeline_type, float)

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

Simular componentes del pipeline con 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)

Pruebas unitarias con DataFrames

def test_transformed_data(clean_data):
    # Incluye otros assert aquí
    ...

    # Comprueba el número de columnas
    assert len(clean_data.columns) == 4

    # Comprueba el límite inferior de una columna
    assert clean_data["open"].min() >= 0

    # Comprueba el rango de una columna encadenando con "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

¡Vamos a practicar!

ETL y ELT en Python