Automating Data Cleaning Pipelines

# CHAPTER 16

Automating Data Cleaning Pipelines #

1. Chapter Introduction #

2. The ETL Paradigm #

ETL stands for Extract, Transform, Load. It is the backbone of data engineering.

1. 1. Extract: Pull raw data from a source (CSV, API, SQL Database).

1. 2. Transform: This is Data Cleaning! (Handle NaNs, duplicates, formats, encodings).

1. 3. Load: Push the cleaned data to a destination (Data Warehouse, clean CSV).

3. Transitioning from Notebooks to Scripts #

The Notebook Way (Messy):

• Variables are scattered.

• Cells can be run out of order.

• Hard to run on a schedule (Cron).

The Script Way (Professional):

• Use a .py file.

• Wrap code in functions.

• Use if _name == "main_": to execute the pipeline.

• Use logging instead of print().

4. Building a Modular Cleaning Pipeline #

Here is a complete, automated pipeline template in Python.

import pandas as pd
import numpy as np
import logging

# 1. Setup Logging (Better than print statements for automation)
logging.basicConfig(
    filename='cleaning_pipeline.log', 
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

# 2. EXTRACT
def extract_data(filepath):
    """Loads raw data."""
    logging.info(f"Extracting data from {filepath}...")
    try:
        df = pd.read_csv(filepath)
        logging.info(f"Successfully loaded {len(df)} rows.")
        return df
    except Exception as e:
        logging.error(f"Failed to extract data: {e}")
        raise

# 3. TRANSFORM (The Cleaning Logic)
def clean_names(df):
    logging.info("Cleaning names...")
    df['name'] = df['name'].str.strip().str.title()
    return df

def handle_missing(df):
    logging.info("Handling missing values...")
    # Drop rows missing critical IDs
    df = df.dropna(subset=['id'])
    # Impute numeric
    if 'salary' in df.columns:
        df['salary'] = df['salary'].fillna(df['salary'].median())
    return df

def validate_data(df):
    logging.info("Validating data constraints...")
    assert df['id'].duplicated().sum() == 0, "Duplicate IDs found!"
    assert df['salary'].min() >= 0, "Negative salaries found!"
    return df

def transform_data(df):
    """Orchestrates the cleaning steps using method chaining / piping."""
    logging.info("Starting transformation phase...")
    df_clean = (df
                .pipe(clean_names)
                .pipe(handle_missing)
                .pipe(validate_data)
               )
    return df_clean

# 4. LOAD
def load_data(df, output_path):
    """Saves the cleaned data."""
    logging.info(f"Loading data to {output_path}...")
    df.to_csv(output_path, index=False)
    logging.info("Pipeline completed successfully.")

# 5. ORCHESTRATOR
def run_pipeline(input_file, output_file):
    try:
        raw_df = extract_data(input_file)
        clean_df = transform_data(raw_df)
        load_data(clean_df, output_file)
    except Exception as e:
        logging.critical(f"PIPELINE FAILED: {e}")

# This ensures the script only runs when executed directly
if __name__ == "__main__":
    INPUT = 'raw_daily_sales.csv'
    OUTPUT = 'clean_daily_sales.csv'
    run_pipeline(INPUT, OUTPUT)

5. Running and Scheduling the Pipeline #

Once your logic is encapsulated in cleaner_pipeline.py, you can run it from the command line:

python cleaner_pipeline.py

Scheduling (Automation):

• Mac/Linux: Use cron to schedule the script.

• Windows: Use Task Scheduler to trigger the python script daily.

• Enterprise: Use workflow orchestrators like Apache Airflow or GitHub Actions.

6. Batch Processing (Handling Multiple Files) #

If you have a folder of 30 daily CSVs, automate the pipeline to loop through them.

import os
import glob

def process_batch_folder(folder_path):
    # Find all CSVs in the folder
    files = glob.glob(os.path.join(folder_path, "*.csv"))
    
    for file in files:
        output_name = file.replace('.csv', '_cleaned.csv')
        print(f"Processing {file}...")
        run_pipeline(file, output_name)

7. Common Mistakes #

• Hardcoding file paths: Avoid df = pd.readcsv('C:/Users/Dave/data.csv'). If someone else runs your script, it crashes. Use relative paths or pass paths as arguments to your functions.

9. Interview Questions #

• Q: How do you transition a messy exploratory Jupyter Notebook into a production-ready ETL script?

• Q: Your automated cleaning pipeline failed at 2:00 AM. How do you design your script so that you can diagnose the problem the next morning?

10. Summary #

11. Next Chapter Recommendation #