Batch Processing Optimizer

Batch Processing Optimizer

Optimize batch data processing workloads using Spark, Polars, DuckDB, and pandas with focus on memory efficiency, parallelism, and cost reduction.

Activation Triggers

Activate on: "batch processing", "Spark optimization", "Polars", "DuckDB", "pandas performance", "data frame", "shuffle optimization", "partition skew", "memory optimization", "out of memory"

NOT for: Real-time streaming → streaming-pipeline-architect | Warehouse SQL tuning → data-warehouse-optimizer | Pipeline orchestration → airflow-dag-orchestrator

Quick Start

1. Choose the right tool — DuckDB for single-node analytics, Polars for DataFrames, Spark for distributed
2. Profile first — identify bottlenecks (shuffle, skew, memory) before optimizing
3. Reduce data early — filter and select columns as early as possible in the pipeline
4. Avoid shuffles — broadcast small tables, pre-partition data, use map-side joins
5. Right-size resources — match executor memory/cores to actual data size

Core Capabilities

Architecture Patterns

Tool Selection Decision Tree

Data Size?
  ├─ < 10 GB     → DuckDB (SQL) or Polars (DataFrame)
  │                 Single machine, zero setup, fastest iteration
  │
  ├─ 10-100 GB   → Polars (lazy) or DuckDB (out-of-core)
  │                 Still single machine with spill-to-disk
  │
  └─ > 100 GB    → Spark (distributed)
                    Multi-node cluster, shuffle-based joins

Complexity?
  ├─ SQL-centric  → DuckDB (fastest SQL engine for analytics)
  ├─ DataFrame    → Polars (10x faster than pandas, lazy evaluation)
  └─ Complex ML   → Spark + MLlib or Spark + Ray

Spark Optimization Patterns

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .getOrCreate()

# GOOD: broadcast small dimension table (< 100MB)
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_dim_df), "key")

# GOOD: predicate pushdown — filter before join
orders = spark.read.parquet("s3://data/orders/") \
    .filter(F.col("order_date") >= "2026-01-01") \
    .select("order_id", "customer_id", "amount")  # column pruning

# BAD: collect() on large dataset — causes OOM on driver
# all_data = large_df.collect()  # NEVER do this

# GOOD: write partitioned output
result.repartition(200) \
    .write.mode("overwrite") \
    .partitionBy("order_date") \
    .parquet("s3://output/results/")

Polars Lazy Evaluation

import polars as pl

# Lazy mode: builds query plan, optimizes, then executes
result = (
    pl.scan_parquet("data/orders/*.parquet")  # lazy scan
    .filter(pl.col("order_date") >= "2026-01-01")
    .join(
        pl.scan_parquet("data/customers/*.parquet"),
        on="customer_id",
        how="inner"
    )
    .group_by("region")
    .agg([
        pl.col("amount").sum().alias("total_revenue"),
        pl.col("order_id").n_unique().alias("order_count"),
    ])
    .sort("total_revenue", descending=True)
    .collect()  # executes optimized plan
)

# Polars optimizes: predicate pushdown, projection pushdown,
# join reordering — all automatically via lazy evaluation

Anti-Patterns

1. pandas for >5GB — pandas loads everything into memory; use Polars (lazy) or DuckDB for medium data, Spark for large
2. Collect to driver — df.collect() or df.toPandas() on large Spark DataFrames causes OOM; aggregate first
3. Ignoring partition skew — one partition with 10x more data than others bottlenecks the entire job; use AQE or salting
4. Reading all columns — always select only needed columns; Parquet columnar format skips unused columns entirely
5. Tiny output files — too many small output files (< 128MB) slow downstream reads; coalesce before writing

Quality Checklist

• Tool matches data size (DuckDB/Polars < 100GB, Spark > 100GB)
• Columns pruned early (select only what is needed)
• Filters pushed down to scan level (predicate pushdown)
• Small tables broadcast in joins (< 100MB)
• Spark AQE enabled (adaptive query execution)
• No collect() on large datasets (aggregate before collecting)
• Output files sized 128MB-1GB (coalesce/repartition before write)
• Partition skew monitored and mitigated (salting or AQE)
• Job profiled: Spark UI stages, Polars .explain(), DuckDB EXPLAIN ANALYZE
• Memory sized appropriately: executor memory >= 2x largest partition