Kiểm thử đơn vị cho pipeline dữ liệu

ETL và ELT với Python

Jake Roach

Data Engineer

Xác thực pipeline dữ liệu với unit tests

Unit tests:

Dùng phổ biến trong quy trình kỹ thuật phần mềm
Đảm bảo mã chạy đúng như kỳ vọng
Giúp xác thực dữ liệu

Khung xác thực pipeline dữ liệu điển hình, gồm kiểm thử đơn vị và kiểm thử end-to-end.

pytest cho kiểm thử đơn vị

from pipeline import extract, transform, load

# Xây dựng unit test, kiểm tra kiểu của 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 và isinstance

pipeline_type = "ETL"

# Kiểm tra pipeline_type có phải là str không
isinstance(pipeline_type, str)

True

# Khẳng định pipeline thực sự có giá trị "ETL"
assert pipeline_type == "ETL"

# Kết hợp assert và isinstance
assert isinstance(pipeline_type, str)

AssertionError

pipeline_type = "ETL"

# Tạo một AssertionError
assert isinstance(pipeline_type, float)

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

Giả lập thành phần pipeline với 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)

Kiểm thử đơn vị DataFrame

def test_transformed_data(clean_data):
    # Thêm các câu lệnh assert khác tại đây
    ...

    # Kiểm tra số cột
    assert len(clean_data.columns) == 4

    # Kiểm tra cận dưới của một cột
    assert clean_data["open"].min() >= 0

    # Kiểm tra phạm vi của một cột bằng cách nối với "and"
    assert clean_data["open"].min() >= 0 and clean_data["open"].max() <= 1000

Cùng luyện tập nào!

ETL và ELT với Python