Strutturare i passaggi di trasformazione come funzioni
Suddivida il notebook in funzioni di estrazione, trasformazione e caricamento, ognuna delle quali accetta e restituisce un DataFrame per facilitarne il test.
Strutturare i passaggi di trasformazione come funzioni è una lezione Pandas & NumPy Academy gratuita su CoddyKit. Questa è la lezione 1 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento Pandas & NumPy Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso Pandas & NumPy Academy include 4 lezioni in totale.
Oltre i notebook
I notebook Jupyter sono ottimi per l'esplorazione, ma poco adatti alle pipeline di dati in produzione. Un codice distribuito in 50 celle, con stato globale e senza test, è fragile: modificare una cella può compromettere silenziosamente un'altra. L'alternativa professionale consiste nell'estrarre ogni trasformazione in una funzione denominata che accetta un DataFrame e restituisce un DataFrame. Questa separazione delle responsabilità è il fondamento di un codice di data engineering facile da mantenere.
# 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'))Il pattern ETL: Extract, Transform, Load
Il pattern ETL suddivide una pipeline di dati in tre fasi: Extract (lettura dalla sorgente), Transform (pulizia e arricchimento) e Load (scrittura nella destinazione). Ogni fase è una funzione separata. Questa separazione facilita la sostituzione delle sorgenti dati (CSV o database), la modifica della logica di pulizia o il cambiamento del formato di output senza intervenire sulle altre due fasi. Ogni pipeline di produzione dovrebbe seguire questa struttura.
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.')Principio di responsabilità singola
Ogni funzione di trasformazione dovrebbe fare esattamente una cosa. Una funzione chiamata clean_data() che elimina i valori nulli, limita gli outlier, analizza le date e codifica le categorie è difficile da testare e da sottoporre a debug. Scriva invece drop_nulls(), cap_outliers(), parse_dates() e encode_categories() come funzioni separate. Questa granularità facilita l'omissione, la sostituzione o il riordinamento di ogni singolo passaggio.
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
)Concatenazione dei passaggi con pipe()
Colleghi le funzioni a responsabilità singola usando pipe() per creare l'intera fase di trasformazione come una catena leggibile. La catena segue la logica di una ricetta: ogni riga è un passaggio e il flusso dei dati procede dall'alto verso il basso. È possibile commentare o riordinare qualsiasi passaggio senza rinominare le variabili. Il risultato finale è il DataFrame pulito e arricchito, pronto per la fase di caricamento.
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)Restituzione del numero di righe per il controllo
Ogni funzione di trasformazione dovrebbe registrare facoltativamente quante righe ha ricevuto e quante ne ha restituite. Racchiudere il corpo della funzione in una registrazione del numero di righe prima e dopo l'elaborazione crea una traccia di controllo leggera, che consente di individuare rapidamente dove vengono eliminate righe durante un'esecuzione della pipeline. Memorizzi questi conteggi in un elenco e li stampi come report al termine di ogni esecuzione.
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)Funzioni testabili con piccoli DataFrame
Il vantaggio principale delle funzioni denominate è la testabilità. Scriva un piccolo DataFrame di test che rappresenti un caso limite realistico e verifichi l'output della funzione. Testi la funzione drop_null_rows con una riga contenente un valore nullo e verifichi che venga eliminata, quindi con un'altra riga senza valori nulli e verifichi che venga mantenuta. Gli unit test sulle funzioni di trasformazione rilevano le regressioni quando cambia il codice della pipeline.
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()Organizzazione delle funzioni in moduli
Man mano che la pipeline cresce, suddivida le funzioni in file di moduli Python: extract.py, transform.py, load.py e validate.py. Lo script principale pipeline.py li importa e li orchestra. Questa struttura di file è leggibile, testabile con pytest e distribuibile come pacchetto Python. Rispecchia l'organizzazione standard adottata dai team di data engineering che usano strumenti come dbt o 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)')Idempotenza: esecuzione sicura più volte
Una funzione di pipeline ben progettata è idempotente: eseguirla due volte sullo stesso input produce lo stesso output e non causa effetti collaterali. Eviti le modifiche in-place (df.drop(..., inplace=True)) e usi sempre df.copy() all'inizio delle funzioni che modificano le colonne. Le funzioni idempotenti possono essere rieseguite dopo un errore senza corrompere i dati di output.
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.')Annotazioni di tipo per l'autodocumentazione
Aggiunga le annotazioni di tipo Python alle firme delle funzioni di trasformazione per rendere esplicito il contratto: def transform(df: pd.DataFrame) -> pd.DataFrame. Le annotazioni di tipo sono autodocumentanti: chiunque legga la funzione sa esattamente cosa si aspetta e cosa restituisce senza doverne leggere il corpo. Consentono inoltre agli strumenti di analisi statica come mypy e al completamento automatico dell'IDE di rilevare gli errori di tipo prima dell'esecuzione.
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.')Documentazione di ogni passaggio con le docstring
Ogni funzione di trasformazione dovrebbe avere una docstring di una riga che indichi cosa fa, quali colonne richiede e quali colonne aggiunge o rimuove. Le docstring ben scritte rendono la funzione individuabile tramite help() e nei suggerimenti dell'IDE. Fungono inoltre da documenti di specifica: un'asserzione non soddisfatta che contraddice la docstring indica un bug; una docstring in disaccordo con il codice indica un errore nella documentazione.
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)Esecuzione della pipeline completa
Orchestri l'ETL completa richiamando in sequenza le tre fasi: extract, transform e load. Aggiunga la misurazione dei tempi intorno a ogni fase per capire dove viene impiegato il tempo. Gestisca separatamente le eccezioni di ogni fase, così che i messaggi di errore identifichino quella che ha avuto esito negativo. Registri l'ora di inizio e di fine dell'intera esecuzione e indichi se è terminata con successo o con un errore, ai fini del monitoraggio.
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}')Verifica rapida
Verifichi la Sua comprensione dei concetti di analisi dei dati trattati in questa lezione.
Riepilogo della lezione
In questa lezione ha imparato a: strutturare le pipeline come funzioni ETL a responsabilità singola, rendere le funzioni testabili, idempotenti e autodocumentanti con annotazioni di tipo e docstring e orchestrare la pipeline completa con misurazione dei tempi e logging di controllo. Nel prossimo argomento esploreremo come parametrizzare le pipeline con dizionari di configurazione per riutilizzarle con dataset diversi.
Domande Frequenti
La lezione «Strutturare i passaggi di trasformazione come funzioni» è gratuita?
Sì — il testo completo di «Strutturare i passaggi di trasformazione come funzioni» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso Pandas & NumPy Academy, passa a CoddyKit PRO. Il corso Pandas & NumPy Academy include 4 lezioni in totale.
Cosa imparerò in «Strutturare i passaggi di trasformazione come funzioni»?
Suddivida il notebook in funzioni di estrazione, trasformazione e caricamento, ognuna delle quali accetta e restituisce un DataFrame per facilitarne il test. Eserciti Pandas & NumPy Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare Pandas & NumPy Academy?
Non è richiesta alcuna esperienza precedente. Pandas & NumPy Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 1 di 4.
Quanto tempo richiede la lezione «Strutturare i passaggi di trasformazione come funzioni»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione Pandas & NumPy Academy?
Sì. Ogni lezione Pandas & NumPy Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Strutturare i passaggi di trasformazione come funzioni
- Parametrizzare le pipeline con dict di configurazione
- Testare i passaggi della pipeline con asserzioni
- Pianificare e registrare le esecuzioni della pipeline