Testes de unidade em um pipeline de dados

ETL e ELT em Python

Jake Roach

Data Engineer

Validando um pipeline de dados com testes de unidade

Testes de unidade:

Muito usados em fluxos de engenharia de software
Garantem que o código funcione como esperado
Ajudam a validar dados

Framework típico de validação de pipeline de dados, incluindo testes de unidade e de ponta a ponta.

pytest para testes de unidade

from pipeline import extract, transform, load

# Construa um teste de unidade, verificando o 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"

# Verifique se pipeline_type é instância de str
isinstance(pipeline_type, str)

True

# Afirme que o pipeline realmente tem valor "ETL"
assert pipeline_type == "ETL"

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

AssertionError

pipeline_type = "ETL"

# Crie um AssertionError
assert isinstance(pipeline_type, float)

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

Simulando componentes do pipeline com 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)

Testando DataFrames com unit tests

def test_transformed_data(clean_data):
    # Inclua outros asserts aqui
    ...

    # Verifique o número de colunas
    assert len(clean_data.columns) == 4

    # Verifique o limite inferior de uma coluna
    assert clean_data["open"].min() >= 0

    # Verifique o intervalo de uma coluna encadeando com "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

Vamos praticar!

ETL e ELT em Python