Automation and Orchestration

Welcome back. It’s the first of November — one month before the deadline — and October’s orders just arrived. Your pipeline already ingests, validates, cleans, and transforms raw CSVs into actionable insights, but you still run Jupyter Notebook cells by hand in the right order. Human-driven workflows work, but they’re error-prone and hard to scale. In this lesson we’ll cover:

Risks of a manual pipeline
Benefits of automation
How an orchestrator sequences and tracks pipeline steps
A simple, modular pipeline pattern you can run locally or schedule in production

Initially you packaged everything into a single Python script and ran it like this:

The image features a person with a KodeKloud t-shirt standing next to a description of tasks related to data pipelines, accompanied by an animated cat character.

python pipeline.py

That made execution deterministic (all steps run every time) and reduced mistakes, but the file quickly became hard to read and maintain. Console output for the single-file run might look like:

Schema validation passed.
Saved orders data to: insights/2025/10/orders.csv
Logged ingestion to logs/ingest_log.csv
Removed 1 rows with missing values in required fields: [1184]
Removed 2 rows with invalid quantity: [1191, 1199]
Removed 2 duplicate rows: []
Removed 4 rows with invalid customer_id: [1165, 1175, 1176, 1198]
Removed 1 rows with invalid product_id: [1201]
Saved 10 dropped rows to: insights/2025/10/orders_dropped.csv
Cleaned data saved to: insights/2025/10/orders_clean.csv

Order Calculations (first 10 rows):
 order_id  customer_name     product_name           quantity  price  line_total
 1154      Chewbacca         Blue Milk Latte           4      4.5      18.0
 1155      Darth Vader       Blue Milk Latte           1      4.5       4.5
 1156      Darth Vader       Blue Milk Latte           1      4.5       4.5
 1157      Leia Organa       Death Star Espresso       3      3.0       9.0
 1158      Han Solo          Tatooine Mocha            1      6.0       6.0
 1159      Padmé Amidala     Wookiee Cappuccino        4      4.5      18.0
 1160      Padmé Amidala     Death Star Espresso       1      3.0       3.0
 1161      Obi-Wan Kenobi    Tatooine Mocha            1      4.0       4.0

Top 3 Products by Revenue:
 product_name         total_revenue
 Blue Milk Latte           103.5
 Wookiee Cappuccino         73.5
 Tatooine Mocha             72.0

Top 3 Customers by Spend:
 customer_name    total_spend
 Chewbacca           62.0
 Leia Organa         59.5
 Darth Vader         53.0

root@host01 ~/code via 🐍 v3.12.3 (venv) $

Splitting into modules To improve readability and reuse, break the single script into focused modules:

ingest.py — read and validate raw file, archive it, and log ingestion
clean.py — drop or correct invalid rows and save cleaned data
transform.py — join cleaned orders to products and customers, compute metrics

Start by centralizing folder and path setup (either in each module or a shared helper). Example setup inside ingest.py to find an orders file:

# ingest.py (setup and find file)
import os
import pandas as pd
import shutil
from datetime import datetime

data_folder = "data"
archive_folder = os.path.join(data_folder, "archive")
insights_folder = "insights"
logs_folder = "logs"
log_path = os.path.join(logs_folder, "ingest_log.csv")

for folder in [data_folder, archive_folder, insights_folder, logs_folder]:
    os.makedirs(folder, exist_ok=True)

print("✅ Folder structure and paths set up.")

# Find an orders file
files = os.listdir(data_folder)
file_name = next((f for f in files if "orders" in f and f.endswith(".csv")), None)

Design each module as an importable function (e.g., run) that:

Accepts explicit parameters (file name, folder paths, etc.)
Returns outputs the caller needs (usually an output_folder path) This makes dependencies explicit and facilitates unit testing.

# Example signature to use in ingest.py, clean.py, transform.py
def run(file_name: str, data_folder: str, archive_folder: str, insights_folder: str, log_path: str):
    """
    Ingest stage should:
      - Read the CSV
      - Validate schema
      - Save a copy to insights/<year>/<month>/
      - Move raw file to archive/
      - Append or update the ingest log
      - Return output folder path for downstream stages
    """
    file_path = os.path.join(data_folder, file_name)
    # ... validate and process ...
    return os.path.join(insights_folder, "2025", "10")

A simple orchestrator script Create run_pipeline.py that imports each module and calls their run functions in order. This file orchestrates the end-to-end flow for a single file:

# run_pipeline.py
import os
from pipeline import ingest, clean, transform

data_folder = "data"
archive_folder = os.path.join(data_folder, "archive")
insights_folder = "insights"
logs_folder = "logs"

products_path = os.path.join(data_folder, "products.csv")
customers_path = os.path.join(data_folder, "customers.csv")
log_path = os.path.join(logs_folder, "ingest_log.csv")

def run_pipeline():
    # Find a new orders file to process
    files = os.listdir(data_folder)
    file_name = next((f for f in files if "orders" in f and f.endswith(".csv")), None)

    if not file_name:
        print("No orders file found.")
        return

    print(f"📦 Starting pipeline for: {file_name}")

    try:
        output_folder = ingest.run(file_name, data_folder, archive_folder, insights_folder, log_path)
        clean.run(products_path, customers_path, output_folder)
        transform.run(products_path, customers_path, output_folder)
        print(f"✔️ Pipeline completed for: {file_name}")
    except Exception as e:
        print(f"✖️ Pipeline failed: {e}")

if __name__ == "__main__":
    run_pipeline()

How modules coordinate Each module receives the specific inputs it needs and returns outputs for downstream stages. The call sequence inside run_pipeline.py becomes explicit and easy to follow:

# Example call sequence inside run_pipeline()
output_folder = ingest.run(file_name, data_folder, archive_folder, insights_folder, log_path)
clean.run(products_path, customers_path, output_folder)
transform.run(products_path, customers_path, output_folder)

This pattern makes reasoning about data flow and side effects easier and supports testing modules independently. Idempotency and state tracking To avoid reprocessing the same file, maintain an ingest log or state file that records processing status for each file (e.g., ingested, cleaned, transformed). This allows resuming a failed pipeline without duplicating work.

Use an ingest log or state-tracking file to make your pipeline idempotent: the orchestrator can check previous state and skip already-completed steps.

A simple CSV ingest log is often enough for small pipelines. Example ingest_log.csv:

file_name,status,rows,timestamp
orders_2025_06.csv,Success,3,2025-05-24T08:00:00
orders_2025_07.csv,Success,30,2025-09-29T11:41:07
orders_2025_08.csv,Success,75,2025-09-29T11:42:36
orders_2025_09.csv,Success,50,2025-09-30T03:15:23
orders_2025_10.csv,Success,50,2025-10-27T08:47:49

Using this safety mechanism, the pipeline is safe to run multiple times without double-processing files.

Scheduling and orchestration Right now you still manually trigger the pipeline:

python run_pipeline.py

For recurring automation, you have several options depending on scale and requirements.

Simple server scheduling: use cron on Unix-like systems.
Cloud scheduler: use AWS EventBridge (or similar) to trigger cloud jobs.
Full workflow orchestration: use Apache Airflow or Prefect for complex dependencies, retries, monitoring, and team collaboration.

Cron example: edit your crontab with crontab -e and add a line to run the script at midnight on the first of every month (replace with your interpreter and absolute path):

0 0 1 * * /usr/bin/python3 /home/youruser/code/run_pipeline.py

Cron schedule fields are: minute, hour, day of month, month, day of week. For monthly runs at midnight on the first day:

0 0 1 * *

The image shows a diagram explaining time units like seconds and minutes, and a person on the right wearing a "KodeKloud" shirt gesturing while talking.

When to use an orchestrator

Use cron or EventBridge for simple periodic tasks (backups, monthly reports).
Use Airflow, Prefect, or other orchestrators when you need:
- Directed acyclic graphs (DAGs) of dependent tasks
- Retry policies and alerting
- Visibility and centralized logs
- Parallel task execution and resource management

The image shows a person wearing a "KodeKloud" shirt standing next to an illustration of a person typing on a laptop, with logos of Apache Airflow and Prefect, and labeled sections such as "More Steps," "Dependencies," and "Teams Involved."

Running the pipeline (example) To process October’s orders, run:

python run_pipeline.py

A successful run prints step-by-step logs and outcomes:

root@host01 ~/code via 🐍 v3.12.3 (venv) ➔ python run_pipeline.py
File not found in log - proceeding.
Starting pipeline for: orders_2025_10.csv
File not ingested before - proceed to next step.
Schema validation passed.
Saved orders data to: insights/2025_10/orders_ingested.csv
Moved raw file to archive/orders_2025_10.csv
Logged ingestion to logs/ingest_log.csv

Cleaning started...
Removed 1 rows with missing values: [1184]
Removed 2 rows with invalid quantity: [1191, 1199]
Saved 10 dropped rows to: insights/2025_10/orders_dropped.csv
Cleaned data saved to: insights/2025_10/orders_clean.csv

Top 3 Products by Revenue:
    product_name         total_revenue
0   Blue Milk Latte           103.5
1   Wookiee Cappuccino         73.5
2   Tatooine Mocha             72.0

Top 3 Customers by Spend:
    customer_name            total_spend
0   Chewbacca Kashyyyk           63.6
1   Leia Organa Alderaan         59.5
2   Darth Vader Mustafar         53.5

Pipeline completed for: orders_2025_10.csv
root@host01 ~/code via 🐍 v3.12.3 (venv) ➔

If the pipeline is re-run, the ingest log prevents duplication:

root@host01 ~/code via 🐍 v3.12.3 (venv) ➔ python run_pipeline.py
File 'orders_2025_10.csv' has already been ingested - skipping.

Quick reference: module responsibilities

Module	Responsibility	Example output
`ingest.py`	Read CSV, validate, archive raw file, update ingest log	`insights/2025/10/orders_ingested.csv`
`clean.py`	Remove/correct invalid rows, save dropped rows	`insights/2025/10/orders_clean.csv`, `orders_dropped.csv`
`transform.py`	Join with `products.csv`/`customers.csv`, compute KPIs	Top products/customers reports

Recap

Manual pipelines are fragile: they depend on human sequencing and can cause mistakes.
Modular automation makes pipelines readable, testable, and maintainable.
Use an ingest log or state file to make your pipeline idempotent and resumable.
Scheduler/orchestrator choice depends on complexity: cron/EventBridge for simple schedules; Airflow/Prefect for DAGs, retries, monitoring, and team workflows.

Next: complete the hands-on exercise to refactor your single-file pipeline into modular stages and add a simple ingest log.

Automation and Orchestration

Watch Video

Practice Lab