0Pricing
Pandas & NumPy Academy · Lesson

Scheduling and Logging Pipeline Runs

Run your pipeline as a Python script from the command line, log start and end times, and use cron or a scheduler for automation.

Scheduling and Logging Pipeline Runs is a free Pandas & NumPy Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the Pandas & NumPy Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

From Notebook to Script

A pipeline that runs only when a developer manually opens a notebook provides no business value beyond the first run. To run automatically every day, the pipeline must be structured as a Python script that is executable from the command line: python pipeline.py. This requires a if __name__ == '__main__': entry point, command-line argument parsing, and proper logging — the three pillars of a production script.

# 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)

Configuring Python Logging

Python's built-in logging module is the correct tool for pipeline logs — not print() statements. Configure a logger with both console output and file output using logging.basicConfig(). Log at the INFO level for normal progress and ERROR for failures. File-based logs persist after the process exits, which is essential for debugging scheduled runs that no one was watching.

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.')

Logging Pipeline Start and End

Always log the start time, end time, and elapsed time of a pipeline run. This establishes a baseline: if the pipeline normally runs in 45 seconds and today it took 8 minutes, something changed — perhaps the input file is 10× larger or a database query is running slowly. Time-stamped start and end log entries make this comparison trivial from the log file alone.

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

Logging Row Counts Per Step

Log the row count entering and exiting each transformation step. A clean log looks like: extract: 50,000 rows → drop_nulls: 49,200 rows → filter: 47,800 rows → output: 47,800 rows. This trace makes it immediately clear how many rows were dropped at each step and whether the numbers are expected. Anomalous drops show up as gaps in the logged counts.

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.')

Scheduling with cron on Linux/Mac

cron is the standard Unix scheduler for recurring jobs. Edit the crontab with crontab -e and add a line specifying when to run the script. The format is: minute hour day month weekday command. A pipeline that must run at 6:00 AM every day uses 0 6 * * * /usr/bin/python /path/to/pipeline.py. Always use absolute paths in cron entries because cron runs in a minimal environment without your shell's PATH settings.

# 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')

Scheduling with Python schedule Library

The schedule library provides a pure-Python way to run jobs at specified intervals without touching cron. It is useful in environments where cron is not available (Windows) or when you want the scheduler logic inside the Python process itself. Wrap the pipeline in a scheduled job loop and keep the process alive to execute repeatedly.

# 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')

Error Handling and Exit Codes

A pipeline script should return a non-zero exit code when it fails so the scheduler knows the job failed. Wrap the main execution in a try/except block and call sys.exit(1) on failure. cron, Jenkins, and Airflow all check the exit code: a non-zero code triggers an alert, re-run, or notification. An unhandled exception that does not set the exit code may go unnoticed by automated monitoring.

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')

Writing a Pipeline Run Summary File

After a successful run, write a small JSON summary file alongside the output. Include the run timestamp, input row count, output row count, rows dropped, and elapsed time. Monitoring systems and dashboards can read this file to track pipeline health trends over time. A dashboard showing output rows over the last 30 days makes it easy to spot the day a data source started delivering fewer records.

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.')

Idempotent Scheduling: Avoiding Double Runs

If a scheduled pipeline is triggered twice accidentally, it should not corrupt the output. Design the load step to be idempotent: use a dated output file name, or overwrite the same output with the latest result. For database loads, use if_exists='replace' or an UPSERT pattern. Never use append mode without a deduplication step, or each scheduled run will add duplicate rows to the output table.

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}')

Alerting on Pipeline Failure

For pipelines that business operations depend on, silence after a failure is dangerous. Set up a simple alert: if the run summary file is not updated within the expected window, send an email or Slack message. Python's smtplib can send an email on failure, or you can use a webhook to post to Slack. Alert immediately on exit code 1 or 2 so the analyst knows the daily refresh missed before the business notices.

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.')

Complete Scheduled Pipeline Script

Combine all the pieces — argument parsing, logging config, run summary, error handling, and exit codes — into a complete pipeline script. This script can be dropped into any environment, pointed at a config file, and scheduled with cron or any workflow orchestrator. It produces a dated log file, a run summary, and a dated output file with every execution, making every run fully auditable and independently reproducible.

# 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')

Quick Check

Test your understanding of Data Analysis concepts from this lesson.

Lesson Recap

In this lesson you learned: structuring a pipeline as a command-line script with argument parsing and logging, scheduling with cron and handling failures with non-zero exit codes and alerts, and writing run summary files and designing idempotent load steps for reliable automated execution. Congratulations on completing the Data Analysis: Pandas and NumPy track!

Frequently asked questions

Is the “Scheduling and Logging Pipeline Runs” lesson free?

Yes — the full text of “Scheduling and Logging Pipeline Runs” is free to read here on the web, and the Pandas & NumPy Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the Pandas & NumPy Academy course, upgrade to CoddyKit PRO.

What will I learn in “Scheduling and Logging Pipeline Runs”?

Run your pipeline as a Python script from the command line, log start and end times, and use cron or a scheduler for automation. You practise Pandas & NumPy Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start Pandas & NumPy Academy?

No prior experience is required. Pandas & NumPy Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Scheduling and Logging Pipeline Runs” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this Pandas & NumPy Academy lesson?

Yes. Every Pandas & NumPy Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Structuring Transformation Steps as Functions
  2. Parameterising Pipelines with Config Dicts
  3. Testing Pipeline Steps with Assertions
  4. Scheduling and Logging Pipeline Runs
← Back to Pandas & NumPy Academy