Enhetstesta en datapipeline

ETL och ELT i Python

Jake Roach

Data Engineer

Validera en datapipeline med enhetstester

Enhetstester:

Vanliga inom mjukvaruutveckling
Säkerställer att koden fungerar som förväntat
Hjälper till att validera data

Typiskt valideringsramverk för datapipelines, som inkluderar enhetstestning och end-to-end-testning.

pytest för enhetstestning

from pipeline import extract, transform, load

# Bygg ett enhetstest som kontrollerar typen för 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 och isinstance

pipeline_type = "ETL"

# Kontrollera om pipeline_type är en instans av str
isinstance(pipeline_type, str)

True

# Kontrollera att pipelinen faktiskt har värdet "ETL"
assert pipeline_type == "ETL"

# Kombinera assert och isinstance
assert isinstance(pipeline_type, str)

AssertionError

pipeline_type = "ETL"

# Skapa ett AssertionError
assert isinstance(pipeline_type, float)

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

Mocka komponenter i datapipelines med 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)

Enhetstesta DataFrames

def test_transformed_data(clean_data):
    # Inkludera andra assert-satser här
    ...

    # Kontrollera antal kolumner
    assert len(clean_data.columns) == 4

    # Kontrollera nedre gränsen för en kolumn
    assert clean_data["open"].min() >= 0

    # Kontrollera intervallet för en kolumn genom att kedja satser med "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

Nu kör vi en övning!

ETL och ELT i Python