Unit test untuk pipeline data

ETL dan ELT di Python

Jake Roach

Data Engineer

Validasi pipeline data dengan unit test

Unit test:

Umum di alur kerja rekayasa perangkat lunak
Memastikan kode berjalan sesuai harapan
Membantu validasi data

Kerangka validasi pipeline data: unit testing dan end-to-end testing.

pytest untuk unit test

from pipeline import extract, transform, load

# Bangun unit test, memastikan tipe 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 dan isinstance

pipeline_type = "ETL"

# Periksa apakah pipeline_type adalah instance str
isinstance(pipeline_type, str)

True

# Pastikan pipeline memang bernilai "ETL"
assert pipeline_type == "ETL"

# Gabungkan assert dan isinstance
assert isinstance(pipeline_type, str)

AssertionError

pipeline_type = "ETL"

# Buat AssertionError
assert isinstance(pipeline_type, float)

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

Mock komponen pipeline data dengan 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)

Unit test untuk DataFrame

def test_transformed_data(clean_data):
    # Sertakan pernyataan assert lain di sini
    ...

    # Periksa jumlah kolom
    assert len(clean_data.columns) == 4

    # Periksa batas bawah kolom
    assert clean_data["open"].min() >= 0

    # Periksa rentang kolom dengan menggabungkan pernyataan dengan "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

Ayo berlatih!

ETL dan ELT di Python