การทำ Unit Test ให้กับ Data Pipeline

ETL and ELT ใน Python

Jake Roach

Data Engineer

การตรวจสอบ Data Pipeline ด้วย Unit Tests

Unit tests:

ใช้กันทั่วไปในกระบวนการทำงานด้าน Software Engineering
ช่วยให้มั่นใจว่าโค้ดทำงานตามที่คาดไว้
ช่วยตรวจสอบความถูกต้องของข้อมูล

กรอบการตรวจสอบ Data Pipeline ทั่วไป ซึ่งรวมถึงการทำ Unit Testing และ End-to-end Testing

pytest สำหรับการทำ Unit Testing

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 and 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

การ Mock องค์ประกอบของ Data Pipeline ด้วย 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)

การทำ Unit Testing กับ DataFrames

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

มาฝึกกันเถอะ!

ETL and ELT ใน Python