Pandas & NumPy Academy · Leçon

Planifier et journaliser les exécutions du pipeline

Exécutez votre pipeline comme script Python depuis la ligne de commande, journalisez les heures de début et de fin et utilisez cron ou un ordonnanceur pour l’automatisation.

Leçon 4 sur 413 étapes

Planifier et journaliser les exécutions du pipeline est une leçon Pandas & NumPy Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage Pandas & NumPy Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours Pandas & NumPy Academy comprend 4 leçons au total.

Du notebook au script

Un pipeline qui ne s'exécute que lorsqu'un développeur ouvre manuellement un notebook n'apporte aucune valeur métier au-delà de sa première exécution. Pour s'exécuter automatiquement chaque jour, le pipeline doit être structuré comme un script Python exécutable depuis la ligne de commande : python pipeline.py. Cela nécessite un point d'entrée if __name__ == '__main__':, l'analyse des arguments de la ligne de commande et une journalisation correcte — les trois piliers d'un script destiné à la production.

# pipeline.py
import argparse
import logging
import pandas as pd

def main(config_path):
    logging.info(f'Starting pipeline with config: {config_path}')
    # ... run ETL steps ...
    logging.info('Pipeline complete.')

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--config', default='config.json')
    args = parser.parse_args()
    main(args.config)

Configurer la journalisation Python

Le module intégré logging de Python est l'outil approprié pour les journaux du pipeline, et non les instructions print(). Configurez un enregistreur avec une sortie à la fois dans la console et dans un fichier à l'aide de logging.basicConfig(). Utilisez le niveau INFO pour la progression normale et ERROR en cas d'échec. Les journaux enregistrés dans des fichiers persistent après l'arrêt du processus, ce qui est essentiel pour déboguer les exécutions planifiées que personne ne surveillait.

import logging
from datetime import date

log_file = f'pipeline_{date.today()}.log'

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s',
    handlers=[
        logging.FileHandler(log_file),
        logging.StreamHandler()
    ]
)

logging.info('Logger configured.')

Journaliser le début et la fin du pipeline

Journalisez toujours l'heure de début, l'heure de fin et la durée d'une exécution du pipeline. Cela établit une référence : si le pipeline s'exécute normalement en 45 secondes mais qu'il lui a fallu 8 minutes aujourd'hui, quelque chose a changé — le fichier d'entrée est peut-être 10 fois plus volumineux ou une requête de base de données s'exécute lentement. Des entrées de journal horodatées au début et à la fin rendent cette comparaison très simple à partir du seul fichier journal.

import time
import logging

def run_pipeline(config):
    start = time.time()
    logging.info(f'Pipeline START | env={config.get("env", "dev")} | input={config["input_path"]}')

    try:
        df = extract(config)
        df_clean = transform(df, config)
        load(df_clean, config)
        elapsed = time.time() - start
        logging.info(f'Pipeline SUCCESS | rows={len(df_clean)} | elapsed={elapsed:.1f}s')
    except Exception as e:
        logging.error(f'Pipeline FAILED | error={e}', exc_info=True)
        raise

Journaliser le nombre de lignes à chaque étape

Journalisez le nombre de lignes à l'entrée et à la sortie de chaque étape de transformation. Un journal clair ressemble à ceci : extraction : 50 000 lignes → suppression_des_nulles : 49 200 lignes → filtrage : 47 800 lignes → sortie : 47 800 lignes. Cette trace montre immédiatement combien de lignes ont été supprimées à chaque étape et si les nombres sont conformes aux attentes. Les suppressions anormales apparaissent sous forme d'écarts dans les nombres journalisés.

def log_step(df, step_name):
    logging.info(f'{step_name}: {len(df):,} rows')
    return df

import pandas as pd

df = (pd.read_csv('orders.csv')
    .pipe(log_step, 'extract')
    .dropna(subset=['revenue'])
    .pipe(log_step, 'drop_nulls')
    .query('quantity > 0')
    .pipe(log_step, 'filter_qty')
)
print('Step logging complete.')

Planifier avec cron sous Linux/Mac

cron est le planificateur Unix standard pour les tâches récurrentes. Modifiez la crontab avec crontab -e et ajoutez une ligne indiquant quand exécuter le script. Le format est le suivant : minute heure jour mois jour_de_la_semaine commande. Un pipeline qui doit s'exécuter tous les jours à 6 h 00 utilise 0 6 * * * /usr/bin/python /path/to/pipeline.py. Utilisez toujours des chemins absolus dans les entrées cron, car cron s'exécute dans un environnement minimal dépourvu des paramètres PATH de votre interpréteur de commandes.

# crontab entry — edit with: crontab -e
# Run pipeline.py at 06:00 every day
# 0 6 * * * /opt/homebrew/bin/python /Users/analyst/pipeline.py --config /Users/analyst/config.json >> /Users/analyst/cron.log 2>&1

# Common cron patterns:
# 0 6 * * *      — daily at 06:00
# 0 */4 * * *    — every 4 hours
# 0 9 * * 1      — every Monday at 09:00
print('Cron schedule format: minute hour day month weekday')

Planifier avec la bibliothèque Python schedule

La bibliothèque schedule permet d'exécuter des tâches à des intervalles définis, entièrement en Python, sans utiliser cron. Elle est utile dans les environnements où cron n'est pas disponible (Windows) ou lorsque vous souhaitez intégrer la logique de planification directement au processus Python. Encapsulez le pipeline dans une boucle de tâche planifiée et maintenez le processus actif afin qu'il s'exécute à plusieurs reprises.

# pip install schedule
# import schedule, time

# def job():
#     logging.info('Scheduled run starting...')
#     run_pipeline(CONFIG)

# schedule.every().day.at('06:00').do(job)
# schedule.every(4).hours.do(job)

# while True:
#     schedule.run_pending()
#     time.sleep(60)

print('schedule library: use for in-process Python scheduling')

Gestion des erreurs et codes de sortie

Un script de pipeline doit renvoyer un code de sortie différent de zéro lorsqu'il échoue, afin que le planificateur sache que la tâche a échoué. Placez l'exécution principale dans un bloc try/except et appelez sys.exit(1) en cas d'échec. cron, Jenkins et Airflow vérifient tous le code de sortie : un code différent de zéro déclenche une alerte, une nouvelle exécution ou une notification. Une exception non gérée qui ne définit pas le code de sortie peut passer inaperçue pour la surveillance automatisée.

import sys

def main():
    try:
        run_pipeline(CONFIG)
        sys.exit(0)  # success
    except AssertionError as e:
        logging.error(f'Data validation failed: {e}')
        sys.exit(2)  # data error
    except Exception as e:
        logging.error(f'Unexpected error: {e}', exc_info=True)
        sys.exit(1)  # general failure

print('Exit code 0=success, 1=error, 2=data failure')

Écrire un fichier récapitulatif d'exécution du pipeline

Après une exécution réussie, écrivez un petit fichier récapitulatif JSON à côté de la sortie. Incluez l'horodatage de l'exécution, le nombre de lignes en entrée, le nombre de lignes en sortie, le nombre de lignes supprimées et la durée. Les systèmes de surveillance et les tableaux de bord peuvent lire ce fichier pour suivre l'évolution de l'état du pipeline au fil du temps. Un tableau de bord affichant le nombre de lignes en sortie au cours des 30 derniers jours permet de repérer facilement le jour où une source de données a commencé à fournir moins d'enregistrements.

import json
from datetime import datetime

def write_run_summary(config, input_rows, output_rows, elapsed):
    summary = {
        'run_at': datetime.now().isoformat(),
        'input_path': config['input_path'],
        'input_rows': input_rows,
        'output_rows': output_rows,
        'rows_dropped': input_rows - output_rows,
        'elapsed_seconds': round(elapsed, 2),
        'status': 'success'
    }
    with open('last_run_summary.json', 'w') as f:
        json.dump(summary, f, indent=2)
    print('Run summary written.')

Planification idempotente : éviter les doubles exécutions

Si un pipeline planifié est déclenché deux fois par erreur, il ne doit pas corrompre la sortie. Concevez l'étape de chargement de manière idempotente : utilisez un nom de fichier de sortie daté ou remplacez la même sortie par le résultat le plus récent. Pour les chargements en base de données, utilisez if_exists='replace' ou un modèle UPSERT. N'utilisez jamais le mode append sans étape de déduplication, sinon chaque exécution planifiée ajoutera des lignes en double à la table de sortie.

from datetime import date

def load_idempotent(df, config):
    # Date-stamped output: each run overwrites its own day's file
    output_path = f"output_{date.today().strftime('%Y%m%d')}.parquet"
    df.to_parquet(output_path, index=False)
    logging.info(f'Loaded {len(df)} rows to {output_path}')

Alerter en cas d’échec du pipeline

Pour les pipelines dont dépendent les opérations de l’entreprise, le silence après un échec est dangereux. Mettez en place une alerte simple : si le fichier récapitulatif de l’exécution n’est pas mis à jour dans le délai prévu, envoyez un e-mail ou un message Slack. Le smtplib de Python peut envoyer un e-mail en cas d’échec, ou vous pouvez utiliser un webhook pour publier un message sur Slack. Alertez immédiatement en cas de code de sortie 1 ou 2 afin que l’analyste sache que l’actualisation quotidienne a échoué avant que l’entreprise ne s’en aperçoive.

import smtplib

def send_failure_alert(error_msg):
    # Example: send plain-text email via SMTP
    # server = smtplib.SMTP('smtp.example.com', 587)
    # server.sendmail('pipeline@company.com',
    #                 'analyst@company.com',
    #                 f'Subject: Pipeline Failed\n\n{error_msg}')
    # server.quit()
    print(f'[ALERT] Would send failure notification: {error_msg}')

# In main():
# except Exception as e:
#     send_failure_alert(str(e))
#     sys.exit(1)
print('Alert integration pattern shown above.')

Script de pipeline planifié complet

Réunissez tous les éléments — analyse des arguments, configuration de la journalisation, récapitulatif de l’exécution, gestion des erreurs et codes de sortie — dans un script de pipeline complet. Ce script peut être déployé dans n’importe quel environnement, utiliser un fichier de configuration et être planifié avec cron ou n’importe quel orchestrateur de flux de travail. À chaque exécution, il produit un fichier journal daté, un récapitulatif de l’exécution et un fichier de sortie daté, ce qui rend chaque exécution entièrement auditable et reproductible indépendamment.

# Full script skeleton:
# 1. parse --config argument
# 2. configure logging to file + console
# 3. load JSON config
# 4. validate config
# 5. run extract() -> transform() -> load()
# 6. write run summary JSON
# 7. sys.exit(0) on success, sys.exit(1) on failure

print('Production pipeline script structure complete.')
print('Schedule with: crontab -e  or  python scheduler.py')

Vérification rapide

Testez votre compréhension des concepts d’analyse de données présentés dans cette leçon.

Réponds-y dans la leçon interactive

Récapitulatif de la leçon

Dans cette leçon, vous avez appris à : structurer un pipeline sous forme de script en ligne de commande avec analyse des arguments et journalisation, planifier des exécutions avec cron et gérer les échecs à l’aide de codes de sortie non nuls et d’alertes, et écrire des fichiers récapitulatifs d’exécution et concevoir des étapes de chargement idempotentes pour garantir des exécutions automatisées fiables. Félicitations, vous avez terminé le parcours Analyse de données : Pandas et NumPy !

Gratuit pour commencer

Apprends Python avec un tuteur IA — gratuit

Écris et exécute du vrai code dans ton navigateur, obtiens de l'aide instantanée d'un tuteur IA disponible 24h/24, et reprends là où tu t'es arrêté sur le web ou dans l'app.

Cours: 30
Leçons: 120

Commencer gratuitement Voir le cours complet

Questions Fréquemment Posées

La leçon « Planifier et journaliser les exécutions du pipeline » est-elle gratuite ?

Oui — le texte complet de « Planifier et journaliser les exécutions du pipeline » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours Pandas & NumPy Academy, passe à CoddyKit PRO. Le cours Pandas & NumPy Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Planifier et journaliser les exécutions du pipeline » ?

Exécutez votre pipeline comme script Python depuis la ligne de commande, journalisez les heures de début et de fin et utilisez cron ou un ordonnanceur pour l’automatisation. Tu pratiques Pandas & NumPy Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer Pandas & NumPy Academy ?

Aucune expérience préalable n'est requise. Pandas & NumPy Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Planifier et journaliser les exécutions du pipeline » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon Pandas & NumPy Academy ?

Oui. Chaque leçon Pandas & NumPy Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

Structurer les étapes de transformation en fonctions
Paramétrer les pipelines avec des dictionnaires de configuration
Tester les étapes du pipeline avec des assertions
Planifier et journaliser les exécutions du pipeline

← Retour à Pandas & NumPy Academy