Performance Tuning Guide

This guide covers performance optimization strategies for Spring Batch RS applications.

Quick Wins

Chunk Size

Increase chunk size to 500-1000 for large datasets

Connection Pooling

Configure appropriate pool sizes for database operations

Buffering

Use buffered I/O for file operations

Batch Operations

Leverage bulk inserts and batch APIs

---

Chunk Size Optimization

Understanding Chunk Size

Chunk size determines how many items are processed before writing to the destination.

Small Chunks (10-50):

• Lower memory usage
• More frequent I/O operations
• Better for debugging
• More transaction overhead

Large Chunks (500-1000):

• Higher memory usage
• Fewer I/O operations
• Better throughput
• Fewer transaction commits

Recommended Chunk Sizes by Use Case

Use Case	Recommended Size	Reason
File to File	500-1000	Minimize I/O overhead
File to Database	500-1000	Batch inserts are efficient
Database to File	100-500	Balance query overhead and memory
Database to Database	500-1000	Bulk operations on both sides
Complex Processing	100-200	Manage memory for transformations
Testing/Development	10-50	Easier debugging

Example Configuration

// Small chunks for development
let step = StepBuilder::new("dev-process")
    .chunk::<Input, Output>(10)  // Small for debugging
    .reader(&reader)
    .writer(&writer)
    .build();

// Large chunks for production
let step = StepBuilder::new("prod-process")
    .chunk::<Input, Output>(1000)  // Large for throughput
    .reader(&reader)
    .writer(&writer)
    .build();

---

Database Performance

Connection Pool Configuration

PostgreSQL

use sqlx::postgres::PgPoolOptions;

let pool = PgPoolOptions::new()
    .max_connections(10)          // Pool size
    .min_connections(2)           // Minimum idle connections
    .acquire_timeout(Duration::from_secs(30))
    .idle_timeout(Duration::from_secs(600))
    .max_lifetime(Duration::from_secs(1800))
    .connect("postgres://...").await?;

MySQL

use sqlx::mysql::MySqlPoolOptions;

let pool = MySqlPoolOptions::new()
    .max_connections(10)
    .min_connections(2)
    .connect("mysql://...").await?;

SQLite

use sqlx::sqlite::SqlitePoolOptions;

let pool = SqlitePoolOptions::new()
    .max_connections(5)  // SQLite has lower concurrency
    .connect("sqlite://...").await?;

Pool Size Guidelines:

• Single reader + single writer: 5-10 connections
• Multiple concurrent jobs: 10-20 connections
• High concurrency: 20-50 connections
• Don’t exceed database max_connections limit

Database Reader Optimization

// Use appropriate page size
let reader = RdbcItemReaderBuilder::<Record>::new()
    .postgres(pool)
    .query("SELECT * FROM large_table ORDER BY id")  // ORDER BY for consistency
    .with_page_size(500)  // Match or exceed chunk size
    .build_postgres();

// For very large datasets, add WHERE clause
let reader = RdbcItemReaderBuilder::<Record>::new()
    .postgres(pool)
    .query("SELECT * FROM large_table WHERE created_at > '2024-01-01' ORDER BY id")
    .with_page_size(1000)
    .build_postgres();

Database Writer Optimization

// Use bulk inserts with appropriate chunk size
let writer = RdbcItemWriterBuilder::<Record>::new()
    .postgres(&pool)
    .table("target_table")
    .column("field1", |r: &Record| r.field1.as_str().into())
    .column("field2", |r: &Record| r.field2.as_str().into())
    .column("field3", |r: &Record| r.field3.into())
    .build_postgres();

let step = StepBuilder::new("bulk-insert")
    .chunk::<Record, Record>(1000)  // Large chunks for bulk inserts
    .reader(&reader)
    .writer(&writer)
    .build();

Concurrent Chunk Writes

By default each write call completes its INSERT before returning, so the write phase is a queue of one. with_concurrency(N) lets up to N chunk INSERTs be in flight at once over the connection pool.

let writer = RdbcItemWriterBuilder::<Record>::new()
    .postgres(&pool)
    .table("target_table")
    .column("field1", |r: &Record| r.field1.as_str().into())
    .column("field2", |r: &Record| r.field2.as_str().into())
    .with_concurrency(4)  // up to 4 chunk INSERTs in flight
    .build_postgres();

cargo run --example benchmark_csv_postgres_xml --features csv,xml,rdbc-postgres

Attention

This is opt-in, and it changes observable behaviour. Enable it only if the following three consequences are acceptable:

1. Ordering is not preserved. Chunks are no longer written in the order they were read. Anything relying on insertion order — a serial primary key matching read order, for instance — will break.
2. Errors are delayed by up to N chunks. A failing write surfaces on a later write call, or at close(), not on the call that dispatched it. skip_limit therefore applies with that same delay.
3. A failing close() now fails the step (see below). This is how the final in-flight batch reports its errors.

Scope. Only PostgreSQL and MySQL honour the setting. SQLite serialises writes behind a database-level lock, so build_sqlite() logs a warning and ignores it. SeaORM and MongoDB writers are unaffected.

Measured Impact

1 000 000 rows, chunk 1 000, 8 columns, PostgreSQL 15, pool of 10 connections. The pipeline is the 3-step CSV → PostgreSQL → XML → PostgreSQL benchmark; concurrency applies to the two PostgreSQL write steps.

with_concurrency	step 1 write	step 3 write	wall-clock	throughput
1 (default)	4.7 s	5.0 s	16.7 s	59 934 rec/s
2	1.2 s	0.3 s	7.3 s	136 460 rec/s
4	0.8 s	0.1 s	6.5 s	154 776 rec/s
8	0.7 s	0.1 s	6.2 s	160 075 rec/s

The write phase is what collapses — 4.7 s → 0.7 s on step 1, 5.0 s → 0.1 s on step 3. End-to-end that is 2.7× on the full pipeline (16.7 s → 6.2 s).

Astuce

Start at 4. Gains are strongly sub-linear and saturate there: 1 → 2 halves the wall-clock, 2 → 4 buys 0.8 s, and 4 → 8 buys only 0.3 s while doubling both the error reporting delay and the number of unordered in-flight chunks.

Keep N at or below your connection pool size — extra tasks only queue waiting for a connection. Past the saturation point PostgreSQL’s WAL and index maintenance are the limit, not the client, so the right N depends on your row width and index count. Measure before committing to a value.

Note that the bottleneck moves rather than disappearing: at concurrency ≥ 2 both write steps become read-bound (step 3’s read share goes from 32 % to 92 %). Once writes are concurrent, further tuning belongs on the read side.

Breaking Change in 0.5.0 — a Failing close() Fails the Step

Before 0.5.0, an error from ItemWriter::close() was logged as a warning and discarded, so a step could report Success after its final flush had failed. As of 0.5.0 a failing close() sets StepStatus::WriteError and the step returns Err.

This is what makes concurrent writes safe: the last in-flight batch learns its fate only at close(), and a swallowed error there would mean silent data loss. open() is unchanged and keeps its non-fatal handling.

If a job of yours previously passed with a close error, it will now fail — that failure was always real, it was just not reported.

Index Optimization

Attention

Critical for Performance:

1. Add indexes on columns in WHERE clauses
2. Add indexes on ORDER BY columns (for pagination)
3. Consider composite indexes for multi-column queries
4. Monitor index usage with EXPLAIN ANALYZE
5. Drop indexes before bulk inserts, rebuild after

-- Good: Index on filter and order columns
CREATE INDEX idx_records_created_id ON records(created_at, id);

-- Query uses index effectively
SELECT * FROM records WHERE created_at > '2024-01-01' ORDER BY id;

---

File I/O Performance

Buffered Reading

use std::io::BufReader;
use std::fs::File;

// Inefficient: No buffering
let file = File::open("large.csv")?;
let reader = CsvItemReaderBuilder::new()
    .from_reader(file);  // Direct file access

// Efficient: Buffered I/O
let file = File::open("large.csv")?;
let buffered = BufReader::with_capacity(64 * 1024, file);  // 64KB buffer
let reader = CsvItemReaderBuilder::new()
    .from_reader(buffered);

Buffered Writing

use std::io::BufWriter;

// Efficient: Buffered writing
let file = File::create("output.csv")?;
let buffered = BufWriter::with_capacity(64 * 1024, file);
let writer = CsvItemWriterBuilder::new()
    .from_writer(buffered);

Memory-Mapped Files

For very large files, consider memory-mapped I/O:

use memmap2::Mmap;
use std::fs::File;

let file = File::open("huge.csv")?;
let mmap = unsafe { Mmap::map(&file)? };
let reader = CsvItemReaderBuilder::new()
    .from_reader(&mmap[..]);

---

Memory Management

Monitoring Memory Usage

use sysinfo::{System, SystemExt};

fn log_memory_usage() {
    let mut system = System::new_all();
    system.refresh_all();

    let used = system.used_memory();
    let total = system.total_memory();

    println!("Memory: {:.2} GB / {:.2} GB ({:.1}%)",
        used as f64 / 1_000_000_000.0,
        total as f64 / 1_000_000_000.0,
        (used as f64 / total as f64) * 100.0
    );
}

Memory-Efficient Patterns

1. Clone Only When Necessary:

// Inefficient: Cloning after taking by value
impl ItemProcessor<Record, Record> for MyProcessor {
    fn process(&self, item: Record) -> ItemProcessorResult<Record> {
        let mut owned = item;
        owned.field = transform(owned.field);
        Ok(Some(owned))
    }
}

// Efficient: Transform and move
impl ItemProcessor<Record, Record> for MyProcessor {
    fn process(&self, item: Record) -> ItemProcessorResult<Record> {
        Ok(Some(Record {
            field: transform(item.field),
            ..item
        }))
    }
}

2. Use References for Large Strings:

use std::borrow::Cow;

struct Processor;

impl ItemProcessor<String, String> for Processor {
    fn process(&self, item: String) -> ItemProcessorResult<String> {
        // Use Cow to avoid unnecessary cloning
        let result: Cow<str> = if item.contains("special") {
            Cow::Owned(item.to_uppercase())
        } else {
            Cow::Owned(item)
        };

        Ok(Some(result.into_owned()))
    }
}

3. Stream Processing:

// Process items one at a time without buffering
let step = StepBuilder::new("stream-process")
    .chunk::<Input, Output>(1)  // Process immediately
    .reader(&reader)
    .processor(&processor)
    .writer(&writer)
    .build();

---

Parallel Processing Considerations

Single-Threaded Model

Spring Batch RS uses a single-threaded model per step execution. For parallel processing:

Option 1: Multiple Job Instances

use std::thread;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let handles: Vec<_> = (0..4).map(|i| {
        thread::spawn(move || {
            process_partition(i)
        })
    }).collect();

    for handle in handles {
        handle.join().unwrap()?;
    }

    Ok(())
}

fn process_partition(partition_id: usize) -> Result<(), Box<dyn std::error::Error>> {
    // Each thread processes a partition
    let reader = CsvItemReaderBuilder::new()
        .from_path(format!("data_part_{}.csv", partition_id))?;

    // ... process partition
    Ok(())
}

Option 2: Async Steps with Tokio

use tokio::task;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let tasks = (0..4).map(|i| {
        task::spawn(async move {
            process_partition(i).await
        })
    });

    futures::future::try_join_all(tasks).await?;

    Ok(())
}

---

Monitoring and Profiling

Basic Metrics

use std::time::Instant;

let start = Instant::now();

let mut execution = StepExecution::new("process");
step.execute(&mut execution)?;

let duration = start.elapsed();
let items = execution.read_count();
let throughput = items as f64 / duration.as_secs_f64();

println!("Processed {} items in {:?}", items, duration);
println!("Throughput: {:.2} items/sec", throughput);
println!("Write count: {}", execution.write_count());
println!("Skip count: {}", execution.skip_count());

Reading the per-phase breakdown

Every chunk-oriented step logs where its time went. This is an info! record, so it only appears if a logger is initialised at info level or below (for example RUST_LOG=info with env_logger):

Step 'load-postgres' 42.3s — read 18.1s (43%) | process 2.4s (6%) | write 21.1s (50%) | flush 0.7s (2%)

The same numbers are available programmatically:

reading step metrics

let job = JobBuilder::new().start(&step).build();
job.run()?;

let metrics = job.get_step_execution("load-postgres").unwrap();
println!("{}", metrics.phase_summary());
println!("read: {:?}", metrics.read_duration);

Percentages are relative to the step’s total duration. The four phases do not sum to 100%. The remainder covers the writer’s open() and close() — which is where the CSV, JSON and XML writers perform their final flush — the teardown of each processed chunk, and the framework’s own bookkeeping. A large remainder is itself worth investigating.

Only chunk-oriented steps populate these fields. On a tasklet step every phase reads 0.0s (0%).

Timing is recorded once per chunk, not once per item, so the overhead stays negligible even on multi-million-row jobs.

Profiling with Criterion

[dev-dependencies]
criterion = "0.5"

use criterion::{black_box, criterion_group, criterion_main, Criterion};

fn benchmark_processor(c: &mut Criterion) {
    let processor = MyProcessor::new();

    c.bench_function("process item", |b| {
        b.iter(|| {
            let item = black_box(create_test_item());
            processor.process(&item)
        })
    });
}

criterion_group!(benches, benchmark_processor);
criterion_main!(benches);

Flamegraph Profiling

# Install cargo-flamegraph
cargo install flamegraph

# Run with profiling
cargo flamegraph --bin my-batch-job

# Open flamegraph.svg in browser

---

Real-World Optimization Example

Before Optimization

// Slow: Small chunks, no buffering, unnecessary clones
let reader = CsvItemReaderBuilder::new()
    .from_path("data.csv")?;  // No buffering

let processor = SlowProcessor;  // Clones everything

let writer = RdbcItemWriterBuilder::<Record>::new()
    .postgres(&small_pool)  // 2 connections only
    .table("target")
    .column("field1", |r: &Record| r.field1.as_str().into())
    .build_postgres();

let step = StepBuilder::new("slow")
    .chunk::<Record, Record>(10)  // Tiny chunks
    .reader(&reader)
    .processor(&processor)
    .writer(&writer)
    .build();

// Result: 100 items/sec

After Optimization

use std::io::BufReader;

// Fast: Large chunks, buffered I/O, optimized pool
let file = File::open("data.csv")?;
let buffered = BufReader::with_capacity(64 * 1024, file);
let reader = CsvItemReaderBuilder::new()
    .from_reader(buffered);

let processor = FastProcessor;  // Minimal cloning

let pool = PgPoolOptions::new()
    .max_connections(20)  // Larger pool
    .connect("...").await?;

let writer = RdbcItemWriterBuilder::<Record>::new()
    .postgres(&pool)
    .table("target")
    .column("field1", |r: &Record| r.field1.as_str().into())
    .column("field2", |r: &Record| r.field2.into())
    .build_postgres();

let step = StepBuilder::new("fast")
    .chunk::<Record, Record>(1000)  // Large chunks
    .reader(&reader)
    .processor(&processor)
    .writer(&writer)
    .build();

// Result: 10,000 items/sec (100x improvement!)

---

Performance Checklist

Astuce

Before Production:

• Chunk size >= 500 for large datasets
• Database queries have appropriate indexes
• Connection pool size matches workload
• File I/O uses buffering
• Processor minimizes cloning
• Writer uses bulk operations
• Memory usage monitored
• Benchmarks run on production-like data
• Profiling identifies bottlenecks
• Error handling doesn’t kill performance (use skip_limit)

Troubleshooting Slow Jobs

Symptom: Low Throughput

Check:

1. Chunk size too small
2. Missing database indexes
3. Network latency (database/FTP)
4. Expensive processor operations
5. Small connection pool

Symptom: High Memory Usage

Check:

1. Chunk size too large
2. Processor accumulating state
3. Memory leaks in custom implementations
4. Large strings not using Cow
5. Unbounded collections

Symptom: Slow Database Writes

Check:

1. Not using bulk inserts
2. Individual INSERT statements
3. Transaction overhead (too frequent commits)
4. Missing indexes on target table
5. Constraints/triggers slowing inserts

See Also

• Feature Flags - Enable only what you need
• Error Handling - Fault tolerance patterns
• Examples - Optimized example code