데이터 파이프라인 단위 테스트

Python으로 ETL과 ELT

Jake Roach

Data Engineer

단위 테스트로 데이터 파이프라인 검증

단위 테스트:

소프트웨어 엔지니어링에서 흔히 사용
코드가 예상대로 동작하는지 확인
데이터 검증에 도움

단위 테스트와 종단간 테스트를 포함한 전형적 데이터 파이프라인 검증 프레임워크.

단위 테스트를 위한 pytest

from pipeline import extract, transform, load

# Build a unit test, asserting the type of 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와 isinstance

pipeline_type = "ETL"

# Check if pipeline_type is an instance of a str
isinstance(pipeline_type, str)

True

# Assert that the pipeline does indeed take value "ETL"
assert pipeline_type == "ETL"

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

AssertionError

pipeline_type = "ETL"

# Create an AssertionError
assert isinstance(pipeline_type, float)

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

fixture로 파이프라인 구성요소 모킹

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)

DataFrame 단위 테스트

def test_transformed_data(clean_data):
    # Include other assert statements here
    ...

    # Check number of columns
    assert len(clean_data.columns) == 4

    # Check the lower bound of a column
    assert clean_data["open"].min() >= 0

    # Check the range of a column by chaining statements with "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

연습해 봅시다!

Python으로 ETL과 ELT