Batch Processing Patterns

Overview

Most data pipelines, ETL jobs, and nightly reports still run on batches. Working on bounded chunks instead of an endless stream keeps the logic easy to reason about — the trade-off is that you have to think hard about idempotency, fault tolerance, and observability.

When to Use

Reach for these patterns when:

• Processing large datasets that do not fit in memory. See Retry Logic for handling transient failures.
• Building ETL pipelines for data warehouses
• Generating nightly reports or aggregations
• Migrating data between systems with downtime windows

Skip batch processing when events need sub-minute latency, the dataset fits comfortably in memory, or work arrives as a continuous stream — a streaming framework or a message queue fits those cases better.

Solution

Resilient Batch Pipeline (Python)

import logging
from typing import Callable, List, Iterator

class BatchProcessor:
    def __init__(self, batch_size: int = 1000, max_retries: int = 3):
        self.batch_size = batch_size
        self.max_retries = max_retries
        self.processed = 0
        self.failed = []

    def process(
        self,
        items: Iterator[dict],
        handler: Callable[[List[dict]], None]
    ) -> dict:
        batch = []
        for item in items:
            batch.append(item)
            if len(batch) >= self.batch_size:
                self._execute(batch, handler)
                batch = []

        if batch:
            self._execute(batch, handler)

        return {"processed": self.processed, "failed": len(self.failed)}

    def _execute(self, batch: List[dict], handler: Callable):
        for attempt in range(self.max_retries):
            try:
                handler(batch)
                self.processed += len(batch)
                return
            except Exception as e:
                logging.warning(f"Batch failed (attempt {attempt + 1}): {e}")
                if attempt == self.max_retries - 1:
                    self.failed.extend(batch)

Idempotent Job Tracking (SQL)

CREATE TABLE job_runs (
    job_id VARCHAR(64) PRIMARY KEY,
    started_at TIMESTAMP NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMP,
    status VARCHAR(20) CHECK (status IN ('running', 'completed', 'failed')),
    checksum VARCHAR(64)
);

-- Before starting, check if already completed
SELECT * FROM job_runs WHERE job_id = 'daily_report_2025_01_15' AND status = 'completed';

Explanation

Before trusting a batch pipeline in production, make sure it has three properties:

1. Idempotency: the same job, run twice, has to land on the same result. Use job IDs and checksums to skip already-processed work. See Idempotent API Endpoints for deduplication patterns.
2. Fault tolerance: one bad batch shouldn’t take the whole job down. Implement retry with exponential backoff and a dead-letter queue.
3. Observability: you need to see progress, throughput, and errors while the job runs, so emit metrics for processed items, latency, and failure rates — the structured logging recipe covers the log format.

Chunking strategy: batch size is a trade-off between memory and throughput — go too small and overhead dominates, go too large and you’re flirting with OOM.

Variants

Pattern	Use Case	Trade-off
Chunked processing	Large files, memory limits	Simpler, higher latency
Parallel workers	CPU-bound transformations	Complex, needs coordination
MapReduce	Distributed aggregation	Scales horizontally
Change Data Capture	Incremental sync	Requires source support

Sizing Batches and Handling Backpressure

Start with 100-1000 items per batch and benchmark with real data. For database inserts, 500-2000 rows per batch balances throughput and transaction size; for file processing, 10-50 MB per chunk avoids OOM on 512 MB containers. Monitor memory with resource.getrusage(resource.RUSAGE_SELF).ru_maxrss in Python. Network-bound APIs tolerate larger batches (500-5000) that amortize per-request latency, while CPU-bound transforms do better with smaller batches (100-500) that parallelize cleanly across workers.

When the downstream system cannot keep up, throttle the producer instead of letting the backlog grow. A semaphore caps in-flight batch writes:

from threading import Semaphore

rate_limiter = Semaphore(10)  # max 10 concurrent batch writes

Database connection pool size already bounds write concurrency. In Celery, set worker_prefetch_multiplier = 1 so workers don’t over-fetch; in Kafka consumers, tune max.poll.records. Watch downstream latency — once it passes roughly 2x baseline, shrink the batch size or pause the producer.

Partitioning and data skew

Split large jobs by a partition key so workers run in parallel: by date (df['created_at'].dt.strftime('%Y-%m-%d')), by hash (df['id'] % num_partitions), or by range (WHERE id BETWEEN 0 AND 99999). Hive-style layouts (data/year=2025/month=01/day=15/data.parquet) work with Spark, Presto, and Athena, and PostgreSQL declarative partitioning (CREATE TABLE orders PARTITION BY RANGE (created_at)) gives you the same split inside the database.

Skew appears when one partition dwarfs the rest — check with Counter(df['partition_key']).most_common(10); if the largest is 10x the median, you have a straggler. Salt the hot key (key + '_' + str(random.randint(0, 9))), process the sub-partitions independently, then merge the results.

Reliability: Retries, Checkpoints, and Dead Letters

Retries. Try the whole batch again first; drop to per-item retry only for records that keep failing. Exponential backoff with jitter prevents a thundering herd of simultaneous retries, and tenacity handles both cleanly:

import random
from tenacity import retry, stop_after_attempt, wait_exponential, wait_random

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=60) + wait_random(0, 2)
)
def process_batch(batch):
    ...

In Celery, use autoretry_for = (ConnectionError,) with retry_backoff = True on the task. Log every retry with the batch ID so operators can tell transient noise from a failing dependency.

Checkpoints. A six-hour job that fails at 5:55 should resume, not restart. Persist progress after each batch:

CREATE TABLE job_checkpoints (
    job_id VARCHAR(64),
    batch_number INT,
    status VARCHAR(20),
    PRIMARY KEY (job_id, batch_number)
);

On restart, run SELECT MAX(batch_number) FROM job_checkpoints WHERE job_id = 'daily_etl' AND status = 'completed' and resume from the next batch. File-based checkpoints work too — write JSON to S3 or local disk after each batch, and use an atomic rename (write to a temp file, then os.replace) so a crash mid-write cannot corrupt the checkpoint. Distributed jobs can coordinate checkpoints through Redis or etcd.

Dead letters. Isolate permanently failing items for inspection instead of dropping them or crashing the run:

CREATE TABLE dead_letters (
    id SERIAL PRIMARY KEY,
    job_id VARCHAR(64),
    item JSONB,
    error TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);

Store the item, the error, and the batch ID, then reprocess with a separate hourly job once the root cause is fixed. AWS SQS does this natively with RedrivePolicy and maxReceiveCount; in Celery, route failures to a dedicated DLQ exchange.

Incremental Loads and Schema Evolution

Process only what changed since the last run by tracking a high-water mark:

CREATE TABLE job_watermarks (
    job_id VARCHAR(64) PRIMARY KEY,
    last_processed_id BIGINT,
    last_processed_at TIMESTAMP
);

-- Next run: fetch only new rows
SELECT * FROM orders
WHERE id > (SELECT last_processed_id FROM job_watermarks WHERE job_id = 'orders_etl')
ORDER BY id LIMIT 10000;

Timestamp-based marks (updated_at > last_processed_at) work when rows update in place, but they miss hard deletes — logical replication catches those instead (PostgreSQL pgoutput, MySQL binlog via Debezium).

Schemas also drift mid-pipeline. Add a schema_version column (ALTER TABLE orders ADD COLUMN schema_version INT DEFAULT 1), let the job branch on it (transform_v2(record) if record['schema_version'] == 2 else transform_v1(record)), and run both versions in parallel until the migration is verified. Teams on Avro or Protobuf get the same guarantee from a schema registry with backward-compatible evolution; JSON pipelines can version schemas by $id and validate with JSON Schema.

Data Quality, Validation, and Reconciliation

Run validation before the data lands in the target, not after. Great Expectations covers the common checks in a few lines:

batch.expect_column_values_to_not_be_null("id")
batch.expect_column_values_to_be_unique("id")
batch.expect_column_values_to_be_between("amount", 0, 10000)

Quarantine rows that fail (INSERT INTO quarantine SELECT * FROM staging WHERE amount IS NULL OR amount < 0) and emit a per-batch report — {"batch_id": 42, "total": 1000, "valid": 995, "invalid": 5} — with an alert when the invalid rate passes 1%.

Reconcile outputs against the source to catch silent loss. Compare row counts per batch, and sums for financial data:

SELECT (SELECT COUNT(*) FROM source.orders WHERE date = '2025-01-15')
     - (SELECT COUNT(*) FROM target.orders WHERE date = '2025-01-15') AS discrepancy;

Persist results in a reconciliation_results table and alert on any nonzero variance within minutes — a nightly job that loses 2% of rows for a month is much harder to backfill than one caught the same night.

Orchestration and Scheduling

A single-node cron job is fine for simple pipelines — add flock so overlapping runs can’t corrupt each other:

flock -n /tmp/batch.lock python batch_job.py

On Kubernetes, CronJob with concurrencyPolicy: Forbid gives you the same guarantee. Distributed work belongs in a real queue: Celery/Redis provides task routing, retries, and dead-letter handling, while AWS Step Functions and Google Cloud Workflows offer managed state machines without running brokers.

Airflow orchestrates multi-step batch DAGs:

with DAG(
    "daily_etl",
    schedule="@daily",
    default_args={"retries": 3, "retry_delay": timedelta(minutes=5)},
) as dag:
    process_task = PythonOperator(task_id="process", python_callable=process_batch)

Set sla for deadline alerts, use Pools to cap concurrency across DAGs, and map tasks over partitions with TaskGroup and expand. On serverless, keep each invocation inside the timeout (15 minutes on Lambda), return ReportBatchItemFailures so SQS requeues only failed items, and chain batches through the queue rather than running one long job.

Files Larger Than Memory

Stream files row by row instead of loading them whole. csv.DictReader pairs well with more_itertools.chunked:

import csv
from more_itertools import chunked

with open("large.csv") as f:
    for batch in chunked(csv.DictReader(f), 1000):
        process(batch)

Pandas exposes the same pattern with pd.read_csv("large.csv", chunksize=10000), and Node.js streams via csv-parse. Binary or fixed-record files call for a different tool: mmap maps the file into virtual memory, so the OS pages data in on demand instead of loading the whole thing:

import mmap

with open("large.bin", "r+b") as f:
    mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
    for i in range(0, len(mm), 65536):
        process(mm[i:i + 65536])
    mm.close()

For 10+ GB files, split -l 1000000 large.csv chunk_ and process the pieces in parallel across workers.

What Works

• Design for idempotency: Every job must be safely retryable
• Log everything: Job start, end, and every batch outcome
• Use transactions: Wrap batch writes in database transactions
• Monitor queue depth: Alert when pending batches exceed thresholds
• Implement circuit breakers: Stop retrying if downstream is unhealthy

Common Mistakes

1. Not handling partial failures: A batch of 1000 where 1 fails needs individual retry
2. Ignoring memory limits: Loading entire datasets into RAM crashes the process
3. Missing checkpointing: A 6-hour job that fails at 5:55 must restart from scratch
4. Silent data loss: Errors logged but not surfaced to operators
5. No rollback strategy: Failed jobs leave the database in an inconsistent state

Troubleshooting

• Pipeline output does not match expectations: validate input schemas, intermediate states, and row counts at each step.
• Data quality degrades over time: add validation checks and anomaly detection; define SLIs for freshness, completeness, and accuracy.
• Job fails intermittently: look for race conditions, external dependencies, and resource contention; retry with idempotency and bounded backoff.
• Schema changes break consumers: use schema registries and backward-compatible evolution.
• Storage costs grow unexpectedly: audit partition retention, compression, and duplicate copies; archive cold data and set lifecycle policies.

Further Reading

• Apache Airflow documentation — DAG authoring, retries, and scheduling semantics
• Kafka Connect — managed source/sink connectors for batch ingestion
• AWS Lambda quotas — timeout and batch-size limits for serverless jobs
• PostgreSQL table partitioning — declarative partitioning for batch-friendly tables
• Great Expectations — data validation framework used above

See Also

• Runnable companion examples — BatchProcessor implementation, tracking schema, and pytest suite
• ETL Pipelines Guide — end-to-end extract, transform, and load architecture
• Dead Letter Queue Runbook — operating a DLQ in production
• Database Indexing — optimizing query performance for batch reads
• Load Testing — validating batch job performance under load