Pianificare e registrare le esecuzioni della pipeline

Una pipeline che viene eseguita solo quando uno sviluppatore apre manualmente un notebook non offre alcun valore aziendale oltre la prima esecuzione. Per eseguirla automaticamente ogni giorno, la pipeline deve essere strutturata come uno script Python eseguibile dalla riga di comando: python pipeline.py. Questo richiede un punto di ingresso if __name__ == '__main__':, l'analisi degli argomenti della riga di comando e un sistema di logging adeguato: i tre pilastri di uno script pronto per la produzione.

Il modulo integrato logging di Python è lo strumento corretto per i log della pipeline, non le istruzioni print(). Configuri un logger con output sia sulla console sia su file usando logging.basicConfig(). Usi il livello INFO per l'avanzamento normale e ERROR per gli errori. I log salvati su file persistono dopo la terminazione del processo, una caratteristica essenziale per il debug delle esecuzioni pianificate che nessuno sta osservando.

Registri sempre l'ora di inizio, l'ora di fine e il tempo trascorso di un'esecuzione della pipeline. In questo modo stabilisce una baseline: se normalmente la pipeline viene eseguita in 45 secondi e oggi ha impiegato 8 minuti, qualcosa è cambiato, ad esempio il file di input è 10 volte più grande oppure una query al database è lenta. Le voci di log con timestamp relative all'inizio e alla fine rendono il confronto immediato basandosi solo sul file di log.

Registri il numero di righe in ingresso e in uscita per ogni passaggio di trasformazione. Un log chiaro è simile a questo: extract: 50,000 rows → drop_nulls: 49,200 rows → filter: 47,800 rows → output: 47,800 rows. Questa traccia rende immediatamente evidente quante righe sono state eliminate in ogni passaggio e se i numeri sono quelli attesi. Riduzioni anomale emergono come variazioni nei conteggi registrati.

cron è lo scheduler Unix standard per i processi ricorrenti. Modifichi il crontab con crontab -e e aggiunga una riga che specifichi quando eseguire lo script. Il formato è: minuto ora giorno mese giorno-della-settimana comando. Per eseguire una pipeline ogni giorno alle 06:00, usi 0 6 * * * /usr/bin/python /path/to/pipeline.py. Usi sempre percorsi assoluti nelle voci cron, perché cron viene eseguito in un ambiente minimo, privo delle impostazioni PATH della shell.

La libreria schedule offre un modo interamente basato su Python per eseguire processi a intervalli specifici senza usare cron. È utile negli ambienti in cui cron non è disponibile (Windows) o quando desidera che la logica dello scheduler risieda direttamente nel processo Python. Inserisca la pipeline in un ciclo di processi pianificati e mantenga il processo attivo per eseguire ripetutamente il lavoro.

Quando si verifica un errore, uno script di pipeline dovrebbe restituire un codice di uscita diverso da zero, così lo scheduler sa che il processo non è riuscito. Inserisca l'esecuzione principale in un blocco try/except e chiami sys.exit(1) in caso di errore. cron, Jenkins e Airflow controllano tutti il codice di uscita: un codice diverso da zero attiva un avviso, una nuova esecuzione o una notifica. Un'eccezione non gestita che non imposta il codice di uscita potrebbe passare inosservata ai sistemi di monitoraggio automatizzati.

Dopo un'esecuzione riuscita, scriva un piccolo file di riepilogo JSON insieme all'output. Includa il timestamp dell'esecuzione, il numero di righe in ingresso, il numero di righe in uscita, le righe eliminate e il tempo trascorso. I sistemi di monitoraggio e i dashboard possono leggere questo file per seguire nel tempo le tendenze relative allo stato della pipeline. Un dashboard che mostra le righe di output negli ultimi 30 giorni rende facile individuare il giorno in cui una sorgente dati ha iniziato a fornire meno record.

Se una pipeline pianificata viene attivata accidentalmente due volte, non dovrebbe danneggiare l'output. Progetti il passaggio di caricamento in modo idempotente: usi un nome file di output con la data oppure sovrascriva lo stesso output con il risultato più recente. Per i caricamenti nei database, usi if_exists='replace' oppure un approccio UPSERT. Non usi mai la modalità append senza un passaggio di deduplicazione, altrimenti ogni esecuzione pianificata aggiungerà righe duplicate alla tabella di output.

Per le pipeline da cui dipendono le operazioni aziendali, il silenzio dopo un errore è pericoloso. Configuri un avviso semplice: se il file di riepilogo dell'esecuzione non viene aggiornato entro l'intervallo previsto, invii un'e-mail o un messaggio Slack. smtplib di Python può inviare un'e-mail in caso di errore, oppure può usare un webhook per pubblicare un messaggio su Slack. Generi immediatamente un avviso in caso di codice di uscita 1 o 2, così l'analista sa che l'aggiornamento giornaliero non è riuscito prima che se ne accorga l'azienda.

Unisca tutti gli elementi — analisi degli argomenti, configurazione del logging, riepilogo dell'esecuzione, gestione degli errori e codici di uscita — in uno script completo per la pipeline. Questo script può essere inserito in qualsiasi ambiente, collegato a un file di configurazione e pianificato con cron o con qualsiasi orchestratore di workflow. A ogni esecuzione produce un file di log datato, un riepilogo dell'esecuzione e un file di output datato, rendendo ogni esecuzione completamente verificabile e riproducibile in modo indipendente.

In questa lezione ha imparato a: strutturare una pipeline come script da riga di comando con analisi degli argomenti e logging, pianificare l'esecuzione con cron e gestire gli errori tramite codici di uscita diversi da zero e avvisi, nonché scrivere file di riepilogo delle esecuzioni e progettare fasi di caricamento idempotenti per un'esecuzione automatizzata affidabile. Congratulazioni per aver completato il percorso Analisi dei dati: Pandas e NumPy!