Transformationsschritte als Funktionen strukturieren
Teilen Sie Ihr Notebook in Extract-, Transform- und Load-Funktionen auf, die jeweils einen DataFrame annehmen und zurückgeben, damit sie leicht getestet werden können.
Transformationsschritte als Funktionen strukturieren ist eine kostenlose Pandas & NumPy Academy-Lektion auf CoddyKit. Dies ist Lektion 1 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des Pandas & NumPy Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der Pandas & NumPy Academy-Kurs umfasst insgesamt 4 Lektionen.
Über Notebooks hinausgehen
Jupyter-Notebooks eignen sich hervorragend für Explorationen, aber schlecht für produktive Datenpipelines. Über 50 Zellen verstreuter Code mit globalem Zustand und ohne Tests ist fragil: Eine Änderung an einer Zelle kann unbemerkt eine andere beschädigen. Die professionelle Alternative besteht darin, jede Transformation in eine benannte Funktion auszulagern, die einen DataFrame entgegennimmt und einen DataFrame zurückgibt. Diese Trennung von Zuständigkeiten bildet die Grundlage für wartbaren Data-Engineering-Code.
# 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'))Das ETL-Muster: Extract, Transform, Load
Das ETL-Muster unterteilt eine Datenpipeline in drei Phasen: Extract (aus der Quelle lesen), Transform (bereinigen und anreichern) und Load (in das Ziel schreiben). Jede Phase ist eine eigene Funktion. Durch diese Trennung lassen sich Datenquellen (CSV oder Datenbank) einfach austauschen, die Bereinigungslogik ändern oder das Ausgabeformat wechseln, ohne die beiden anderen Phasen anzupassen. Jede produktive Pipeline sollte dieser Struktur folgen.
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.')Single-Responsibility-Prinzip
Jede Transformationsfunktion sollte genau eine Aufgabe erfüllen. Eine Funktion namens clean_data(), die Nullwerte entfernt, Ausreißer begrenzt, Datumsangaben parst und Kategorien codiert, ist schwer zu testen und zu debuggen. Schreiben Sie stattdessen drop_nulls(), cap_outliers(), parse_dates() und encode_categories() als separate Funktionen. Diese Granularität macht es einfach, einzelne Schritte zu überspringen, zu ersetzen oder neu anzuordnen.
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
)Schritte mit pipe() verketten
Verbinden Sie Funktionen mit jeweils einer Zuständigkeit mithilfe von pipe() zu einer übersichtlichen Kette für die gesamte Transformationsphase. Die Kette liest sich wie ein Rezept: Jede Zeile ist ein Schritt, und der Datenfluss verläuft von oben nach unten. Jeder Schritt kann auskommentiert oder neu angeordnet werden, ohne Variablen umbenennen zu müssen. Das Endergebnis ist der bereinigte und angereicherte DataFrame, der für die Ladephase bereitsteht.
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)Zeilenanzahlen für Audits zurückgeben
Jede Transformationsfunktion sollte optional protokollieren, wie viele Zeilen sie erhalten und wie viele sie zurückgegeben hat. Ein Protokoll der Zeilenanzahl vor und nach der Verarbeitung bietet einen schlanken Audit-Trail, mit dem Sie schnell erkennen, an welcher Stelle eines Pipelinelaufs Zeilen entfernt werden. Speichern Sie diese Anzahlen in einer Liste und geben Sie sie am Ende jedes Laufs als Bericht aus.
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)Testbare Funktionen mit kleinen DataFrames
Der größte Vorteil benannter Funktionen ist ihre Testbarkeit. Erstellen Sie einen kleinen Test-DataFrame, der einen realistischen Sonderfall abbildet, und prüfen Sie die Ausgabe der Funktion. Testen Sie die Funktion drop_null_rows mit einer Zeile, die einen Nullwert enthält, und überprüfen Sie, dass sie entfernt wird; testen Sie außerdem eine Zeile ohne Nullwert und überprüfen Sie, dass sie erhalten bleibt. Unit-Tests für Transformationsfunktionen erkennen Regressionen, wenn sich der Pipeline-Code ändert.
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()Funktionen in Modulen organisieren
Wenn die Pipeline wächst, teilen Sie die Funktionen auf Python-Moduldateien auf: extract.py, transform.py, load.py und validate.py. Das Hauptskript pipeline.py importiert und orchestriert sie. Diese Dateistruktur ist übersichtlich, mit pytest testbar und als Python-Paket bereitstellbar. Sie entspricht dem Standardaufbau, den Data-Engineering-Teams mit Tools wie dbt oder Airflow verwenden.
# 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)')Idempotenz: Mehrfach sicher ausführen
Eine gut konzipierte Pipeline-Funktion ist idempotent: Ihre zweimalige Ausführung mit derselben Eingabe erzeugt dieselbe Ausgabe und verursacht keine Seiteneffekte. Vermeiden Sie Änderungen direkt am Objekt (df.drop(..., inplace=True)) und verwenden Sie am Anfang von Funktionen, die Spalten ändern, immer df.copy(). Idempotente Funktionen können nach Fehlern erneut ausgeführt werden, ohne die Ausgabedaten zu beschädigen.
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.')Typhinweise zur Selbstdokumentation
Fügen Sie den Signaturen von Transformationsfunktionen Python-Typhinweise hinzu, um den Vertrag eindeutig zu machen: def transform(df: pd.DataFrame) -> pd.DataFrame. Typhinweise sind selbstdokumentierend – jede Entwicklerin und jeder Entwickler erkennt beim Lesen der Funktion genau, was sie erwartet und zurückgibt, ohne den Funktionskörper lesen zu müssen. Außerdem ermöglichen sie statischen Analysetools wie mypy und der automatischen Vervollständigung in IDEs, Typfehler bereits vor der Laufzeit zu erkennen.
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.')Jeden Schritt mit Docstrings dokumentieren
Jede Transformationsfunktion sollte über einen einzeiligen Docstring verfügen, der angibt, was sie tut, welche Spalten sie benötigt und welche Spalten sie hinzufügt oder entfernt. Gute Docstrings machen die Funktion über help() und in IDE-Tooltips auffindbar. Sie dienen außerdem als Spezifikationsdokumente: Eine fehlgeschlagene Assertion, die dem Docstring widerspricht, ist ein Fehler; ein Docstring, der nicht mit dem Code übereinstimmt, ist ein Dokumentationsfehler.
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)Die vollständige Pipeline ausführen
Orchestrieren Sie den vollständigen ETL-Prozess, indem Sie die drei Phasen nacheinander aufrufen: Extract, Transform und Load. Fügen Sie um jede Phase eine Zeitmessung ein, um zu ermitteln, wo die Zeit verbraucht wird. Fangen Sie Ausnahmen aus jeder Phase separat ab, damit Fehlermeldungen erkennen lassen, welche Phase fehlgeschlagen ist. Protokollieren Sie für Monitoring-Zwecke Beginn und Ende des gesamten Laufs sowie, ob er erfolgreich war oder fehlgeschlagen ist.
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}')Kurzübung
Testen Sie Ihr Verständnis der Konzepte der Datenanalyse aus dieser Lektion.
Zusammenfassung der Lektion
In dieser Lektion haben Sie gelernt: Pipelines als ETL-Funktionen mit jeweils einer Zuständigkeit zu strukturieren, Funktionen mit Typhinweisen und Docstrings testbar, idempotent und selbstdokumentierend zu machen und die vollständige Pipeline mit Zeitmessung und Audit-Protokollierung zu orchestrieren. Als Nächstes untersuchen wir, wie sich Pipelines mithilfe von Config-Dictionaries für die Wiederverwendung mit unterschiedlichen Datensätzen parametrisieren lassen.
Häufig gestellte Fragen
Ist die Lektion „Transformationsschritte als Funktionen strukturieren“ kostenlos?
Ja — der vollständige Text von „Transformationsschritte als Funktionen strukturieren“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des Pandas & NumPy Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der Pandas & NumPy Academy-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Transformationsschritte als Funktionen strukturieren“?
Teilen Sie Ihr Notebook in Extract-, Transform- und Load-Funktionen auf, die jeweils einen DataFrame annehmen und zurückgeben, damit sie leicht getestet werden können. Du übst Pandas & NumPy Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um Pandas & NumPy Academy zu starten?
Keine Vorkenntnisse erforderlich. Pandas & NumPy Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 1 von 4.
Wie lange dauert die Lektion „Transformationsschritte als Funktionen strukturieren“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser Pandas & NumPy Academy-Lektion Code schreiben und ausführen?
Ja. Jede Pandas & NumPy Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Transformationsschritte als Funktionen strukturieren
- Pipelines mit Config-Dicts parametrisieren
- Pipelineschritte mit Assertions testen
- Pipeline-Läufe planen und protokollieren