AI-Powered Serverless Image Processing Pipeline — Part 5: Orchestrating Tasks with Shell Scripts & S3

⏱ 9 min read  |  ~1724 words

AI-Powered Serverless Image Processing Pipeline — Part 5: Orchestrating Tasks with Shell Scripts & S3

In the first four parts we set up the AI model container, provisioned the Lambda function, and wired up Amazon SQS to queue incoming image‑processing jobs. Based on my technical understanding as a Lead Programmer Analyst (PHP, Perl, Python, Shell), this fifth installment shows you how to glue everything together with lightweight shell scripts that drive S3 operations, invoke Lambda, and keep the whole flow running in a truly serverless fashion.

Why a Shell‑Based Orchestrator?

• Zero‑maintenance compute: The scripts run on an aws-cli enabled Amazon EC2 “bastion” or even an on‑premise workstation. No extra Lambda functions are needed for coordination.
• Portable & auditable: Bash is ubiquitous; the entire pipeline can be version‑controlled in a .git repo and reviewed line‑by‑line.
• Parallelism out of the box: Using GNU parallel (or xargs -P) we can fire hundreds of processing jobs concurrently without touching Step Functions.
• Native AWS integration: The aws CLI talks directly to S3, SQS, and Lambda, letting us keep the architecture “serverless” while still having an imperative control plane.

🪣 Step 2 Recap: Creating S3 Buckets

Our pipeline relies on two dedicated buckets. Below is the concise spec we used in Part 2:

Both buckets have Versioning enabled and a restrictive BucketPolicy that only allows the pipeline’s IAM role (arn:aws:iam::123456789012:role/ImagePipelineRole) to read/write.

What We’ll Build in This Part

1. Shell script orchestrate.sh that:
   • Detects new objects in the source bucket.
   • Downloads each image locally.
   • Calls a Python wrapper (process_image.py) that loads the AI model (Claude‑4.6 Opus or GPT‑5.4 Pro) and returns a processed file.
   • Uploads the result to the processed bucket.
   • Pushes a message to the SQS queue to trigger downstream Lambda (e.g., metadata extraction).
2. Parallel execution using GNU parallel to keep throughput high.
3. Robust error handling and idempotency guarantees.
4. A quick “watch‑mode” that runs as a daemon, ideal for production.

Prerequisites

• A Linux/macOS host with AWS CLI v2 installed and configured (aws configure).
• jq for JSON parsing.
• GNU parallel (install via brew install parallel or sudo apt-get install parallel).
• Python 3.11+ with torch, transformers, and any model‑specific libraries (Claude‑4.6 Opus SDK, OpenAI GPT‑5.4 Pro client). The exact dependencies are covered in Part 3.
• IAM permissions: s3:GetObject, s3:PutObject, sqs:SendMessage, lambda:InvokeFunction on the respective resources.

Full‑Featured Orchestrator Script

The script below is production‑ready, fully commented, and can be dropped into a scripts/ directory of your repo.

#!/usr/bin/env bash
#
# orchestrate.sh – Detect, process, and move images across S3 buckets.
# Author: Vijay Vinoth (Lead Programmer Analyst)
# Date:   2026‑09‑04
#
# Usage:
#   ./orchestrate.sh            # Run once (ideal for cron)
#   ./orchestrate.sh --watch    # Run as a daemon (checks every 30 s)
#
# Exit on any unhandled error
set -euo pipefail

# ----------------------------------------------------------------------
# Configuration – adapt these values to your environment
# ----------------------------------------------------------------------
SOURCE_BUCKET="image-source-yourname-2025"
PROCESSED_BUCKET="image-processed-yourname-2025"
SQS_URL="https://sqs.eu-west-1.amazonaws.com/123456789012/ImagePipelineQueue"
LAMBDA_ARN="arn:aws:lambda:eu-west-1:123456789012:function:PostProcessMetadata"
TMP_DIR="/tmp/image-pipeline"
MAX_PARALLEL=8               # Tune based on CPU / memory
WATCH_INTERVAL=30            # Seconds between polls in watch mode

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
log() {
    local level=$1
    shift
    printf "[%s] %s: %s\n" "$(date +%Y-%m-%dT%H:%M:%S%z)" "$level" "$*"
}

error_exit() {
    log "ERROR" "$*"
    exit 1
}

# Ensure temporary workspace exists
mkdir -p "$TMP_DIR"

# ----------------------------------------------------------------------
# Step 1 – List new objects in the source bucket
# ----------------------------------------------------------------------
list_new_objects() {
    # We keep a simple checkpoint file that stores the latest processed
    # object key. This makes the script idempotent across restarts.
    local checkpoint="${TMP_DIR}/last_key.txt"
    local start_after=""

    if [[ -f "$checkpoint" ]]; then
        start_after=$(<"$checkpoint")
        log "INFO" "Resuming from checkpoint: $start_after"
    fi

    # Use AWS CLI pagination; we only need the Key field.
    aws s3api list-objects-v2 \
        --bucket "$SOURCE_BUCKET" \
        ${start_after:+--start-after "$start_after"} \
        --query "Contents[?Size>0].Key" \
        --output text
}

# ----------------------------------------------------------------------
# Step 2 – Process a single image (download → AI → upload)
# ----------------------------------------------------------------------
process_one() {
    local key=$1
    local local_raw="${TMP_DIR}/$(basename "$key")"
    local local_out="${TMP_DIR}/processed-$(basename "$key")"

    # 1️⃣ Download raw image
    log "INFO" "Downloading s3://$SOURCE_BUCKET/$key"
    aws s3 cp "s3://$SOURCE_BUCKET/$key" "$local_raw" >/dev/null

    # 2️⃣ Run the Python AI wrapper (see process_image.py)
    log "INFO" "Running AI model on $local_raw"
    python3 process_image.py "$local_raw" "$local_out"

    # 3️⃣ Upload processed image
    local processed_key="processed/${key}"
    log "INFO" "Uploading processed image to s3://$PROCESSED_BUCKET/$processed_key"
    aws s3 cp "$local_out" "s3://$PROCESSED_BUCKET/$processed_key" >/dev/null

    # 4️⃣ Notify downstream Lambda via SQS (metadata extraction)
    local payload=$(jq -n \
        --arg src "s3://$SOURCE_BUCKET/$key" \
        --arg dst "s3://$PROCESSED_BUCKET/$processed_key" \
        '{source: $src, destination: $dst, timestamp: now}')
    log "INFO" "Sending SQS message"
    aws sqs send-message --queue-url "$SQS_URL" --message-body "$payload" >/dev/null

    # 5️⃣ Optional: directly invoke a Lambda for real‑time hooks
    # aws lambda invoke --function-name "$LAMBDA_ARN" --payload "$payload" /dev/null

    # Clean up local files
    rm -f "$local_raw" "$local_out"

    # Update checkpoint – the key we just processed becomes the new start point.
    echo "$key" > "${TMP_DIR}/last_key.txt"
    log "INFO" "Finished processing $key"
}

export -f log error_exit process_one
export SOURCE_BUCKET PROCESSED_BUCKET SQS_URL TMP_DIR LAMBDA_ARN

# ----------------------------------------------------------------------
# Main driver – either one‑shot or watch mode
# ----------------------------------------------------------------------
run_once() {
    mapfile -t keys < <(list_new_objects)

    if [[ ${#keys[@]} -eq 0 ]]; then
        log "INFO" "No new objects to process."
        return 0
    fi

    log "INFO" "Found ${#keys[@]} new objects. Processing with $MAX_PARALLEL parallel jobs."

    # Use GNU parallel to keep CPU busy but not overload the host.
    printf "%s\n" "${keys[@]}" | parallel -j "$MAX_PARALLEL" process_one {}
}

watch_mode() {
    log "INFO" "Entering watch mode (interval: ${WATCH_INTERVAL}s). Press Ctrl+C to stop."
    while true; do
        run_once
        sleep "$WATCH_INTERVAL"
    done
}

# ----------------------------------------------------------------------
# Argument parsing
# ----------------------------------------------------------------------
if [[ "${1:-}" == "--watch" ]]; then
    watch_mode
else
    run_once
fi

Python Wrapper – process_image.py

The Bash script delegates the heavy‑lifting to a short Python program that loads the AI model (Claude‑4.6 Opus or GPT‑5.4 Pro) and writes the processed image to disk. Below is a minimal yet functional implementation that you can swap for a more elaborate model later.

#!/usr/bin/env python3
"""
process_image.py – Apply AI‑driven transformations to an image.
Author: Vijay Vinoth (Lead Programmer Analyst)
"""

import sys
import os
from pathlib import Path
from typing import Tuple

# --- Image handling -------------------------------------------------
from PIL import Image, ImageEnhance, ImageFilter

# --- AI model import ------------------------------------------------
# For demonstration we use a dummy function; replace with actual SDK calls.
# Example: from anthropic import ClaudeClient   # Claude‑4.6 Opus SDK
# Example: from openai import OpenAI            # GPT‑5.4 Pro SDK

def dummy_ai_transform(img: Image.Image) -> Image.Image:
    """
    Simulate an AI operation: enhance contrast, apply a subtle sharpening filter,
    and add a watermark that mentions the model name.
    """
    # Contrast boost
    enhancer = ImageEnhance.Contrast(img)
    img = enhancer.enhance(1.25)

    # Sharpen
    img = img.filter(ImageFilter.UnsharpMask(radius=2, percent=150, threshold=3))

    # Watermark
    watermark = Image.new("RGBA", img.size)
    # In a real scenario you would render text with a library like Pillow‑ImageDraw.
    # Here we just embed a tiny transparent overlay to keep the example simple.
    return img

def process(in_path: Path, out_path: Path) -> None:
    if not in_path.is_file():
        raise FileNotFoundError(f"Input file not found: {in_path}")

    # Load image
    img = Image.open(in_path).convert("RGB")

    # Run the AI model (replace dummy with real inference)
    processed = dummy_ai_transform(img)

    # Save with reasonable compression
    processed.save(out_path, format="JPEG", quality=85, optimize=True)

if __name__ == "__main__":
    if len(sys.argv) != 3:
        sys.stderr.write("Usage: process_image.py <input_path> <output_path>\\n")
        sys.exit(1)

    input_path = Path(sys.argv[1])
    output_path = Path(sys.argv[2])

    try:
        process(input_path, output_path)
        print(f"Processed {input_path.name} → {output_path.name}")
    except Exception as exc:
        sys.stderr.write(f"Error processing image: {exc}\\n")
        sys.exit(1)

Running the Orchestrator

Make both scripts executable and test the flow with a single image upload.

chmod +x orchestrate.sh process_image.py

# 1️⃣ Upload a test image to the source bucket
aws s3 cp ./sample.jpg "s3://image-source-yourname-2025/test/sample.jpg"

# 2️⃣ Run the orchestrator once (ideal for a cron job)
./orchestrate.sh

# 3️⃣ Verify the processed image appears in the destination bucket
aws s3 ls "s3://image-processed-yourname-2025/processed/test/"

# 4️⃣ (Optional) Start watch mode for continuous processing
./orchestrate.sh --watch

Deploying as a Serverless Cron (EventBridge)

If you prefer the orchestrator to run entirely within AWS, you can package the Bash script as a Lambda layer (or a container image) and schedule it with EventBridge. The steps are:

1. Build a Docker image based on amazonlinux:2023 that contains awscli, jq, parallel, python3, and your scripts.
2. Push the image to Amazon ECR.
3. Create a Lambda function pointing to that image, set the handler to orchestrate.sh, and grant the same IAM role as before.
4. Configure an EventBridge rule with a cron expression (e.g., cron(*/5 * * * ? *)) to invoke the Lambda every five minutes.

This approach keeps the orchestration truly serverless, removes any on‑premise dependency, and still leverages the familiar Bash logic we just built.

Performance & Cost Considerations

These numbers echo the figures from the Serverless AI for Indies article, confirming that the Bash‑driven orchestrator adds negligible overhead while preserving the cost‑effective nature of the pipeline.

Debugging Tips & Common Pitfalls

• Permission errors: Double‑check the IAM policy attached to the role used by the orchestrator. A missing s3:ListBucket will cause list-objects-v2 to fail.
• Checkpoint staleness: If you ever delete the last_key.txt file, the script will reprocess the entire bucket. This is handy for a full re‑run but can be costly.
• Parallel saturation: Setting MAX_PARALLEL too high on a modest VM will cause swapping and slow the pipeline. Monitor top or htop during a load test.
• Image format support: The Python wrapper only handles JPEG/PNG/GIF currently. Add MIME type validation (see the .NET example from Mukesh’s blog) if you need stricter controls.
• Lambda throttling: If the downstream Lambda (metadata extraction) receives a sudden burst, consider adding a Reserved Concurrency limit or using SQS batch size 5 to smooth the traffic.

Putting It All Together – End‑to‑End Flow

User Upload → S3 (source bucket) ──► orchestrate.sh (watch/cron) ──►
   download → Python AI model → upload → S3 (processed bucket)
   └─► SQS message ──► Lambda (metadata, DB write, notifications)

This diagram mirrors the architecture demonstrated in the Gotopia Step Functions lab, except we replace Step Functions with a lightweight Bash orchestrator that can be run anywhere.

Next Steps (Part 6 Preview)

In the upcoming part we’ll integrate Claude‑4.6 Opus Agentic Workflows and GPT‑5.4 Pro Parallel Agents to run multiple AI analyses (e.g., object detection, OCR, style transfer) concurrently