Strukturyzowanie etapów transformacji jako funkcji
Podziel notebook na funkcje wyodrębniania, transformacji i wczytywania, z których każda przyjmuje i zwraca DataFrame, co ułatwia testowanie.
Strukturyzowanie etapów transformacji jako funkcji to bezpłatna lekcja Pandas & NumPy Academy na CoddyKit. To lekcja 1 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej Pandas & NumPy Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs Pandas & NumPy Academy zawiera 4 lekcji w sumie.
Wyjście poza notatniki
Notatniki Jupyter świetnie sprawdzają się podczas eksploracji, ale słabo nadają się do produkcyjnych potoków danych. Kod rozproszony w 50 komórkach, z globalnym stanem i bez testów, jest podatny na błędy: zmiana jednej komórki może po cichu zepsuć inną. Profesjonalną alternatywą jest wyodrębnienie każdej transformacji do nazwanej funkcji, która przyjmuje DataFrame i zwraca DataFrame. Taki podział odpowiedzialności stanowi podstawę łatwego w utrzymaniu kodu inżynierii danych.
# Notebook-style (fragile)
df = pd.read_csv('orders.csv')
df = df.dropna(subset=['revenue'])
df = df[df['quantity'] > 0]
df['revenue_per_unit'] = df['revenue'] / df['quantity']
# Function-style (robust)
def extract(path):
return pd.read_csv(path)
def transform(df):
return (
df.dropna(subset=['revenue'])
.query('quantity > 0')
.assign(revenue_per_unit=lambda d: d['revenue'] / d['quantity'])
)
df = transform(extract('orders.csv'))Wzorzec ETL: Extract, Transform, Load
Wzorzec ETL dzieli potok danych na trzy fazy: Extract (odczyt ze źródła), Transform (czyszczenie i wzbogacanie) oraz Load (zapis w miejscu docelowym). Każda faza jest osobną funkcją. Taki podział ułatwia zmianę źródeł danych (CSV lub baza danych), modyfikację logiki czyszczenia albo zmianę formatu wyjściowego bez ingerowania w pozostałe fazy. Każdy produkcyjny potok powinien mieć taką strukturę.
def extract(config):
return pd.read_csv(config['input_path'], parse_dates=['order_date'])
def transform(df, config):
return (
df
.dropna(subset=config['required_cols'])
.query('quantity > 0')
.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
)
def load(df, config):
df.to_parquet(config['output_path'], index=False)
print(f'Saved {len(df)} rows.')Zasada pojedynczej odpowiedzialności
Każda funkcja transformacji powinna robić dokładnie jedną rzecz. Funkcja o nazwie clean_data(), która usuwa wartości null, ogranicza wartości odstające, analizuje daty i koduje kategorie, jest trudna do testowania i debugowania. Zamiast tego napisz osobne funkcje drop_nulls(), cap_outliers(), parse_dates() i encode_categories(). Taki poziom szczegółowości ułatwia pominięcie, zastąpienie lub zmianę kolejności dowolnego pojedynczego kroku.
def drop_null_rows(df, required_cols):
return df.dropna(subset=required_cols)
def remove_returns(df):
return df[df['quantity'] > 0]
def compute_revenue(df):
return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
def add_date_features(df):
return df.assign(
year=lambda d: d['order_date'].dt.year,
month=lambda d: d['order_date'].dt.month
)Łączenie kroków za pomocą pipe()
Połącz funkcje o pojedynczej odpowiedzialności za pomocą pipe(), aby zbudować całą fazę transformacji w czytelny łańcuch. Łańcuch przypomina przepis: każda linia jest krokiem, a dane przepływają od góry do dołu. Dowolny krok można zakomentować lub zmienić jego kolejność bez zmiany nazw zmiennych. Końcowym wynikiem jest oczyszczony i wzbogacony DataFrame gotowy do fazy ładowania.
import pandas as pd
REQUIRED = ['order_id', 'revenue', 'order_date']
def transform(raw_df):
return (
raw_df
.pipe(drop_null_rows, required_cols=REQUIRED)
.pipe(remove_returns)
.pipe(compute_revenue)
.pipe(add_date_features)
)
df_clean = transform(pd.read_csv('orders.csv', parse_dates=['order_date']))
print(df_clean.shape)Zwracanie liczby wierszy na potrzeby audytu
Każda funkcja transformacji powinna opcjonalnie rejestrować liczbę otrzymanych i zwróconych wierszy. Umieszczenie w treści funkcji logowania liczby wierszy przed i po jej wykonaniu zapewnia prosty ślad audytowy, który pozwala szybko sprawdzić, w którym miejscu uruchomienia potoku wiersze są usuwane. Przechowuj te liczby na liście i wyświetlaj je w raporcie na końcu każdego uruchomienia.
audit_log = []
def audited(func):
def wrapper(df, *args, **kwargs):
before = len(df)
result = func(df, *args, **kwargs)
after = len(result)
audit_log.append({'step': func.__name__, 'in': before, 'out': after, 'dropped': before - after})
return result
return wrapper
@audited
def drop_null_rows(df, required_cols):
return df.dropna(subset=required_cols)Testowalne funkcje z małymi DataFrame'ami
Największą zaletą nazwanych funkcji jest możliwość ich testowania. Utwórz mały testowy DataFrame reprezentujący realistyczny przypadek brzegowy i sprawdź wynik funkcji. Przetestuj funkcję drop_null_rows z jednym wierszem zawierającym wartość null i upewnij się, że zostanie usunięty, a następnie z drugim, który nie zawiera wartości null, i sprawdź, czy zostanie zachowany. Testy jednostkowe funkcji transformacji wykrywają regresje po zmianach w kodzie potoku.
import pandas as pd
def test_drop_null_rows():
test_df = pd.DataFrame({
'order_id': [1, 2, 3],
'revenue': [100.0, None, 200.0]
})
result = drop_null_rows(test_df, required_cols=['revenue'])
assert len(result) == 2, 'Should have 2 non-null rows'
assert result['revenue'].isna().sum() == 0, 'No nulls in revenue'
print('test_drop_null_rows PASSED')
test_drop_null_rows()Organizowanie funkcji w moduły
W miarę rozrastania się potoku podziel funkcje na pliki modułów Pythona: extract.py, transform.py, load.py i validate.py. Główny skrypt pipeline.py importuje je i koordynuje ich działanie. Taka struktura plików jest czytelna, testowalna za pomocą pytest i możliwa do wdrożenia jako pakiet Pythona. Odzwierciedla standardowy układ używany przez zespoły inżynierii danych korzystające z narzędzi takich jak dbt lub Airflow.
# pipeline.py
# from extract import extract_orders
# from transform import transform
# from load import load_to_parquet
# from validate import validate_sales_df
# def run_pipeline(config):
# raw = extract_orders(config)
# clean = transform(raw, config)
# validate_sales_df(clean)
# load_to_parquet(clean, config)
print('Module-based pipeline structure shown above (imports commented for demo)')Idempotencja: bezpieczne wielokrotne uruchamianie
Dobrze zaprojektowana funkcja potoku jest idempotentna: dwukrotne uruchomienie jej na tych samych danych wejściowych daje ten sam wynik i nie powoduje efektów ubocznych. Unikaj modyfikacji w miejscu (df.drop(..., inplace=True)) i zawsze używaj df.copy() na początku funkcji modyfikujących kolumny. Funkcje idempotentne można ponownie uruchomić po awariach bez ryzyka uszkodzenia danych wyjściowych.
def add_revenue_flag(df, threshold=500):
# Use copy to avoid mutating the input
df = df.copy()
df['is_large_order'] = df['revenue'] >= threshold
return df
# Running twice gives the same result
df1 = add_revenue_flag(df_clean)
df2 = add_revenue_flag(df_clean)
assert df1.equals(df2), 'Function is not idempotent!'
print('Idempotency check passed.')Adnotacje typów jako samodokumentowanie kodu
Dodaj adnotacje typów Pythona do sygnatur funkcji transformacji, aby jasno określić ich kontrakt: def transform(df: pd.DataFrame) -> pd.DataFrame. Adnotacje typów dokumentują się same — każdy programista czytający funkcję wie dokładnie, czego ona oczekuje i co zwraca, bez konieczności czytania jej treści. Umożliwiają także narzędziom analizy statycznej, takim jak mypy, oraz funkcjom autouzupełniania w IDE wykrywanie błędów typów przed uruchomieniem programu.
from typing import List
def drop_null_rows(df: pd.DataFrame, required_cols: List[str]) -> pd.DataFrame:
return df.dropna(subset=required_cols)
def compute_revenue(df: pd.DataFrame) -> pd.DataFrame:
return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
print('Type-annotated functions ready for production use.')Dokumentowanie każdego kroku za pomocą docstringów
Każda funkcja transformacji powinna zawierać jednolinijkowy docstring określający, co robi, jakich kolumn wymaga oraz jakie kolumny dodaje lub usuwa. Dobre docstringi ułatwiają znalezienie funkcji za pomocą help() oraz w podpowiedziach IDE. Służą także jako dokumenty specyfikacji: nieudane sprawdzenie, które jest sprzeczne z docstringiem, oznacza błąd; docstring niezgodny z kodem jest błędem dokumentacji.
def compute_revenue(df: pd.DataFrame) -> pd.DataFrame:
"""Add 'revenue' column as quantity * unit_price.
Requires: 'quantity' (numeric), 'unit_price' (numeric) columns.
Returns: df with new 'revenue' float column appended.
"""
return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
help(compute_revenue)Uruchamianie pełnego potoku
Skoordynuj kompletny proces ETL, wywołując kolejno trzy fazy: extract, transform i load. Dodaj pomiar czasu wokół każdej fazy, aby sprawdzić, na co jest poświęcany czas. Przechwytuj wyjątki z każdej fazy osobno, aby komunikaty o błędach wskazywały, która faza zakończyła się niepowodzeniem. Rejestruj czas rozpoczęcia i zakończenia całego uruchomienia oraz informację, czy zakończyło się ono powodzeniem, czy niepowodzeniem, na potrzeby monitorowania.
import time
CONFIG = {
'input_path': 'orders.csv',
'output_path': 'orders_clean.parquet',
'required_cols': ['order_id', 'revenue']
}
t0 = time.time()
raw = extract(CONFIG)
clean = transform(raw, CONFIG)
load(clean, CONFIG)
print(f'Pipeline completed in {time.time()-t0:.1f}s')
print(f'Audit log: {audit_log}')Szybki test
Sprawdź swoje zrozumienie pojęć związanych z analizą danych z tej lekcji.
Podsumowanie lekcji
W tej lekcji nauczyłeś się: organizować potoki jako funkcje ETL z pojedynczą odpowiedzialnością, tworzyć funkcje testowalne, idempotentne i samodokumentujące się za pomocą adnotacji typów oraz docstringów oraz koordynować działanie pełnego potoku z pomiarem czasu i logowaniem audytowym. Następnie zajmiemy się parametryzowaniem potoków za pomocą słowników konfiguracyjnych, aby można było ponownie używać ich dla różnych zbiorów danych.
Ucz się Python dzięki korepetycjom AI — za darmo
Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.
- Kursy
- 30
- Lekcje
- 120
Często zadawane pytania
Czy lekcja „Strukturyzowanie etapów transformacji jako funkcji” jest bezpłatna?
Tak — pełny tekst „Strukturyzowanie etapów transformacji jako funkcji” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu Pandas & NumPy Academy, przejdź na CoddyKit PRO. Kurs Pandas & NumPy Academy zawiera 4 lekcji w sumie.
Co nauczysz się w „Strukturyzowanie etapów transformacji jako funkcji”?
Podziel notebook na funkcje wyodrębniania, transformacji i wczytywania, z których każda przyjmuje i zwraca DataFrame, co ułatwia testowanie. Ćwiczysz Pandas & NumPy Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć Pandas & NumPy Academy?
Nie wymagamy żadnego doświadczenia. Pandas & NumPy Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 1 z 4.
Ile czasu zajmuje lekcja „Strukturyzowanie etapów transformacji jako funkcji”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji Pandas & NumPy Academy?
Tak. Każda lekcja Pandas & NumPy Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Strukturyzowanie etapów transformacji jako funkcji
- Parametryzowanie potoków za pomocą słowników konfiguracji
- Testowanie etapów potoku za pomocą asercji
- Harmonogramowanie i rejestrowanie uruchomień potoku