Scaling Geospatial Pipelines: From 10 Files to 10,000 in AWS Batch

# Dockerfile hardening for geospatial pipelines
FROM ghcr.io/osgeo/gdal:ubuntu-small-3.9.0

# Prevent Python from writing .pyc files and buffer stdout/stderr
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    GDAL_CACHEMAX=512 \
    CPL_VSIL_CURL_ALLOWED_EXTENSIONS=".tif,.tiff,.vrt,.json,.parquet"

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
ENTRYPOINT ["python", "-u", "-m", "pipeline.worker"]

import rasterio
from rasterio.windows import Window
import numpy as np

def process_raster_in_windows(src_path: str, dst_path: str, block_size: int = 1024):
    """Processes large rasters in bounded memory windows, preventing Exit Code 137."""
    with rasterio.open(src_path) as src:
        profile = src.profile.copy()
        profile.update(
            tiled=True,
            blockxsize=512,
            blockysize=512,
            compress="deflate",
            predictor=2,
            bigtiff="yes"
        )
        
        with rasterio.open(dst_path, "w", **profile) as dst:
            # Iterate through bounded coordinate windows
            for col_off in range(0, src.width, block_size):
                for row_off in range(0, src.height, block_size):
                    window = Window(
                        col_off=col_off,
                        row_off=row_off,
                        width=min(block_size, src.width - col_off),
                        height=min(block_size, src.height - row_off)
                    )
                    # Peak memory bounded to block_size * block_size * bands * dtype
                    data = src.read(window=window).astype(np.float32)
                    
                    # Compute localized geospatial index or inference
                    result = (data[3] - data[2]) / (data[3] + data[2] + 1e-6)
                    
                    dst.write(result.astype(rasterio.float32), 1, window=window)
                    del data, result  # Explicit GC release

// Least-privilege IAM Policy attached to AWS Batch Job Role
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowImageryIngest",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:ListBucket"],
      "Resource": [
        "arn:aws:s3:::sentinel-cogs/*",
        "arn:aws:s3:::sentinel-cogs"
      ]
    },
    {
      "Sid": "AllowProductionCOGPublish",
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:AbortMultipartUpload"],
      "Resource": "arn:aws:s3:::infryne-production-features/*"
    }
  ]
}

# Step 1: Submit single Array Job handle representing 10,000 parallel tiles
aws batch submit-job \
  --job-name sentinel-tile-pipeline-202607 \
  --job-queue geospatial-compute-spot \
  --job-definition tile-processor:4 \
  --array-properties size=10000 \
  --container-overrides '{
    "environment": [
      {"name": "MANIFEST_S3_URI", "value": "s3://infryne-jobs/manifests/2026-run.jsonl"}
    ]
  }'

import signal
import sys
import boto3

class SpotGracefulTerminator:
    def __init__(self, tile_id: str, checkpoint_bucket: str):
        self.tile_id = tile_id
        self.bucket = checkpoint_bucket
        self.interrupted = False
        # Intercept POSIX signals emitted during Spot reclamation
        signal.signal(signal.SIGINT, self.handle_sigterm)
        signal.signal(signal.SIGTERM, self.handle_sigterm)

    def handle_sigterm(self, signum, frame):
        print(f"[SHUTDOWN] Captured signal {signum}. Evacuating state for {self.tile_id}...")
        self.interrupted = True
        self.save_checkpoint_to_s3()
        sys.exit(143)  # Exit code 143 signals Spot interruption to AWS Batch

    def save_checkpoint_to_s3(self):
        s3 = boto3.client("s3")
        s3.put_object(
            Bucket=self.bucket,
            Key=f"checkpoints/{self.tile_id}.state",
            Body=b'{"status": "interrupted", "stage": "inference"}'
        )

"retryStrategy": {
  "attempts": 3,
  "evaluateOnExit": [
    {
      "onStatusReason": "Host EC2 (instance *) terminated.",
      "action": "RETRY"
    },
    {
      "onExitCode": "143",
      "action": "RETRY"
    },
    {
      "onExitCode": "1",
      "action": "EXIT"
    }
  ]
}

import os, time, random
from botocore.config import Config
import boto3

# Bounded startup jitter prevents synchronized thundering herd on S3
jitter_seconds = random.uniform(0.5, 90.0)
time.sleep(jitter_seconds)

# Configure boto3 client with aggressive exponential backoff
boto_config = Config(
    retries={
        "max_attempts": 10,
        "mode": "adaptive"  # Automatically rate-limits based on server 503 responses
    },
    max_pool_connections=50
)
s3_client = boto3.client("s3", config=boto_config)

import rasterio
from rasterio.env import Env

# Set production GDAL virtual file system environment variables
with Env(
    GDAL_DISABLE_READDIR_ON_OPEN="EMPTY_DIR",
    CPL_VSIL_CURL_ALLOWED_EXTENSIONS=".tif",
    VSI_CACHE=True,
    VSI_CACHE_SIZE=67108864,  # 64 MB
    GDAL_HTTP_MERGE_CONSECUTIVE_RANGES="YES"
):
    with rasterio.open("s3://sentinel-cogs/T43RER_20260714.tif") as src:
        # Reads only the required bytes via HTTP Range headers
        ndvi_window = src.read(1, window=Window(0, 0, 512, 512))

import json
import boto3
import rasterio
from rasterio.errors import RasterioIOError

def safe_tile_executor(tile_uri: str, sqs_dlq_url: str):
    try:
        with rasterio.open(tile_uri) as src:
            if src.width == 0 or src.height == 0:
                raise ValueError("Raster has zero spatial extent")
            return process_tile(src)
    except (RasterioIOError, ValueError, Exception) as err:
        print(f"[POISON PILL] Quarantining corrupt tile {tile_uri}: {err}")
        sqs = boto3.client("sqs")
        sqs.send_message(
            QueueUrl=sqs_dlq_url,
            MessageBody=json.dumps({
                "tile_uri": tile_uri,
                "error_type": type(err).__name__,
                "error_message": str(err),
                "timestamp": time.time()
            })
        )
        # Return cleanly so parent batch array job continues processing
        return None