intermediate By Mathias Paulenko Last updated: Jul 9, 2026

Batch Processing Patterns

Q: How do I monitor batch job throughput and progress?

Emit metrics at each batch boundary. In Python, use prometheus_client: from prometheus_client import Counter, Histogram; processed = Counter('batch_processed_total', 'Items processed'); latency = Histogram('batch_duration_seconds', 'Batch duration'). Push metrics to Prometheus Pushgateway for batch jobs: from prometheus_client import push_to_gateway; push_to_gateway('localhost:9091', job='batch_etl', registry=registry). For cloud setups, emit CloudWatch custom metrics or Datadog statsd. Track: items processed, items failed, batch duration, queue depth, and memory usage. Set up Grafana dashboards with alerts for: throughput below 50% of average, failure rate above 5%, and job duration exceeding SLA. Log structured JSON for each batch: {"batch_id": 42, "processed": 1000, "failed": 3, "duration_ms": 1250}.

Design reliable batch processing pipelines for large datasets with retry logic, idempotency, and observability.

batch-processing data performance architecture parsing

Topics: data

Overview

Batch processing is the backbone of data pipelines, ETL workflows, and report generation. Unlike stream processing, batch jobs process bounded datasets in chunks, making them simpler to reason about but requiring careful attention to idempotency, fault tolerance, and observability.

When to Use

Use this resource 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

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

A production batch pipeline needs three properties:

Idempotency: Running the same job twice must produce the same result. Use job IDs and checksums to skip already-processed work. See Idempotent API Endpoints for deduplication patterns.
Fault tolerance: Individual batch failures should not crash the entire job. Implement retry with exponential backoff and a dead-letter queue.
Observability: Track progress, throughput, and errors. Log metrics for processed items, latency, and failure rates.

Chunking strategy: Size batches to balance memory usage and throughput. Too small = overhead; too large = OOM risk.

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

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

Not handling partial failures: A batch of 1000 where 1 fails needs individual retry
Ignoring memory limits: Loading entire datasets into RAM crashes the process
Missing checkpointing: A 6-hour job that fails at 5:55 must restart from scratch
Silent data loss: Errors logged but not surfaced to operators
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 data 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.

Production Notes

Deploy gradually using canary or blue-green to catch regressions early.
Configure alerts for error rate, p99 latency, and failure rate before enabling in production.
Document the rollback in the runbook; test the procedure in staging at least once per quarter.
Review structured logs with correlation IDs to trace requests end-to-end during incidents.

Key Takeaways

Apply batch processing patterns when you need a practical solution for your use case.
Monitor performance after implementation; measure latency, errors, and resource usage before and after.
Check the Troubleshooting section for common failures; most have documented root causes with fixes.
Keep dependencies updated and run tests in CI to prevent production regressions.

Common Production Pitfalls

Copying the example without adapting it to real data volumes and failure modes.
Skipping load and error-injection tests before the first production deployment.
Hard-coding values that should be configurable per environment.
Forgetting to add logging and monitoring at each step.
Deploying without a rollback plan or a tested backup strategy.
Assuming the minimal example will scale without adding caching or batching.
Not documenting the version and configuration used in production.
Letting the recipe sit unchanged when dependencies or scale evolve.

Frequently Asked Questions

How large should each batch be?

Start with 100-1000 items per batch. Benchmark with your data and memory constraints. 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 usage with resource.getrusage(resource.RUSAGE_SELF).ru_maxrss in Python. If batches exceed available memory, reduce batch size or switch to streaming. For network-bound APIs, larger batches (500-5000) amortize latency. For CPU-bound transforms, smaller batches (100-500) allow better parallelism. Always test with production-like data volumes — synthetic data rarely reveals memory pressure patterns.

Should I use a job queue like Celery or a cron job?

Use Celery/Redis for distributed systems with multiple workers, retry logic, and monitoring. Celery provides task routing, priority queues, and dead-letter handling out of the box. For single-node, simple pipelines, cron jobs suffice — but add flock to prevent overlapping runs: flock -n /tmp/batch.lock python batch_job.py. For Kubernetes, use CronJob resources with concurrencyPolicy: Forbid. See Rate Limiting for controlling throughput. For cloud-native setups, AWS Step Functions or Google Cloud Workflows provide managed retry and state management without infrastructure overhead.

How do I handle schema changes mid-pipeline?

Version your job logic and data schemas. Run old and new versions in parallel during migration. Use a schema_version column in target tables: ALTER TABLE orders ADD COLUMN schema_version INT DEFAULT 1. The batch job checks schema_version and applies the correct transformation. For Avro/Protobuf, use schema registry to manage backward-compatible changes. For JSON, validate with JSON Schema versioned by $id. Deploy new job versions with a feature flag: if config.use_new_schema: transform_v2(record) else: transform_v1(record). Monitor both versions' output for discrepancies. Once confident, decommission the old version.

How do I implement checkpointing for long-running batch jobs?

Checkpointing saves progress so failed jobs resume from the last successful batch. Store checkpoints in a durable database table: CREATE TABLE job_checkpoints (job_id VARCHAR(64), batch_number INT, status VARCHAR(20), PRIMARY KEY (job_id, batch_number)). After each batch, write: INSERT INTO job_checkpoints VALUES ('daily_etl', 42, 'completed'). On restart, query the last completed batch: SELECT MAX(batch_number) FROM job_checkpoints WHERE job_id = 'daily_etl' AND status = 'completed'. Resume from batch 43. For file-based checkpoints, write a JSON file to S3 or local disk after each batch. Use atomic writes to prevent corruption: write to a temp file, then rename. For distributed jobs, use Redis or etcd for distributed checkpoint coordination.

How do I handle dead-letter queues for failed batch items?

Dead-letter queues (DLQ) isolate failed items for manual inspection and retry. In Python, maintain a DLQ list: dead_letter_queue = [] and append failed items with context: dead_letter_queue.append({'item': item, 'error': str(e), 'batch_id': batch_id, 'timestamp': datetime.utcnow()}). After the job completes, write the DLQ to a database table or message queue. In Celery, configure task_queues with a dedicated DLQ exchange. In AWS SQS, set RedrivePolicy to move messages after maxReceiveCount attempts. For database-backed DLQs: CREATE TABLE dead_letters (id SERIAL PRIMARY KEY, job_id VARCHAR(64), item JSONB, error TEXT, created_at TIMESTAMP DEFAULT NOW()). Process DLQ items separately with a retry job that runs every hour.

How do I monitor batch job throughput and progress?

Emit metrics at each batch boundary. In Python, use prometheus_client: from prometheus_client import Counter, Histogram; processed = Counter('batch_processed_total', 'Items processed'); latency = Histogram('batch_duration_seconds', 'Batch duration'). Push metrics to Prometheus Pushgateway for batch jobs: from prometheus_client import push_to_gateway; push_to_gateway('localhost:9091', job='batch_etl', registry=registry). For cloud setups, emit CloudWatch custom metrics or Datadog statsd. Track: items processed, items failed, batch duration, queue depth, and memory usage. Set up Grafana dashboards with alerts for: throughput below 50% of average, failure rate above 5%, and job duration exceeding SLA. Log structured JSON for each batch: {"batch_id": 42, "processed": 1000, "failed": 3, "duration_ms": 1250}.

How do I handle backpressure in batch processing pipelines?

Backpressure occurs when downstream systems cannot keep up with the batch producer. Implement rate limiting with a semaphore: from threading import Semaphore; rate_limiter = Semaphore(10) — acquire before each batch write and release after. For database writes, use connection pooling with a max pool size to limit concurrent inserts. For API calls, implement a token bucket: import time; time.sleep(1 / max_rps). Monitor downstream latency — if it increases beyond 2x baseline, reduce batch size or pause processing. In Celery, set worker_prefetch_multiplier = 1 to prevent workers from over-fetching. For Kafka-based pipelines, set max.poll.records to control batch size. Use concurrent.futures.ThreadPoolExecutor(max_workers=N) to limit parallelism in Python.

How do I test batch processing jobs for correctness?

Test batch jobs with deterministic data fixtures. Create a test dataset: test_data = [{'id': i, 'value': i * 2} for i in range(10000)]. Test idempotency by running the job twice and verifying the output is identical: assert run_job(test_data) == run_job(test_data). Test fault tolerance by injecting failures: def failing_handler(batch): if len(batch) > 500: raise Exception('simulated failure') and verify the job retries and records failures. Test checkpointing by killing the job mid-run and verifying it resumes correctly. Use property-based testing with Hypothesis: @given(st.lists(st.dictionaries(st.text(), st.integers()))) to generate edge-case inputs. For SQL-based jobs, use testcontainers to spin up a real database: from testcontainers.postgres import PostgresContainer.

How do I handle batch processing for time-series data?

Time-series batch jobs process data in fixed time windows. Use windowed batching: group records by window_start and window_end timestamps. For example, SELECT date_trunc('hour', timestamp) AS window, COUNT(*) FROM events GROUP BY 1 processes hourly batches. For late-arriving data, use watermarking: allow data up to 5 minutes late by setting watermark = current_time - 5 minutes. Process windows only after the watermark passes. For retention, partition by time: CREATE TABLE events_2025_01 PARTITION OF events FOR VALUES FROM ('2025-01-01') TO ('2025-02-01'). Drop old partitions instead of deleting rows: DROP TABLE events_2024_01. For downsampling, aggregate raw data into 1-minute, 1-hour, and 1-day summary tables in separate batch jobs.

How do I handle batch processing with exactly-once semantics?

Exactly-once processing requires idempotent writes and transactional checkpoints. Use a transaction to write both the batch output and the checkpoint atomically: BEGIN; INSERT INTO results SELECT * FROM staging; INSERT INTO job_checkpoints VALUES ('job_1', 42, 'completed'); COMMIT;. If the transaction fails, both the data and checkpoint are rolled back — the job retries the batch. For Kafka consumers, use enable.idempotence=true and transactional.id for producer-side exactly-once. For database writes, use INSERT ... ON CONFLICT DO NOTHING to handle duplicate batches safely. For external API calls, use idempotency keys: Idempotency-Key: batch_42_run_3 in request headers. Accept that exactly-once has a performance cost — measure whether at-least-once with idempotent writes suffices for your use case.

Recipe

Batch Processing Patterns

Overview

When to Use

Solution

Resilient Batch Pipeline (Python)

Idempotent Job Tracking (SQL)

Explanation

Variants

What Works

Common Mistakes

Troubleshooting

Further Reading

Production Notes

Key Takeaways

See Also

Common Production Pitfalls

Frequently Asked Questions

Caching & Memoization in Python, JavaScript, and Java

Validate and Sanitize User Input Data

Date Formatting

Deep Clone in JavaScript: structuredClone vs lodash vs JSON

Flatten and Unflatten Nested Objects

Deep Clone Objects in JavaScript: Beyond JSON.parse

Overview

When to Use

Solution

Resilient Batch Pipeline (Python)

Idempotent Job Tracking (SQL)

Explanation

Variants

What Works

Common Mistakes

Troubleshooting

Further Reading

Production Notes

Key Takeaways

See Also

Common Production Pitfalls

Frequently Asked Questions

Related Resources

Caching & Memoization in Python, JavaScript, and Java

Validate and Sanitize User Input Data

Date Formatting

Deep Clone in JavaScript: structuredClone vs lodash vs JSON

Flatten and Unflatten Nested Objects

Deep Clone Objects in JavaScript: Beyond JSON.parse