Spring Batch is a powerful framework designed to support the development of robust batch processing applications. It’s a part of the larger Spring ecosystem and is used for processing large volumes of data. Whether you need to execute millions of database records, process large files, or perform other time-consuming operations, Spring Batch provides the necessary tools and infrastructure to make batch processing easier and more manageable.
Part 8: Link
Pre-requisites:
Spring Batch is a fairly advanced concept. Before reading this tutorial, you should know at least the basic Java programming and the fundamental Spring Boot concepts.
Table of Contents:
- Introduction to Batch Processing
- Overview of Spring Batch
- Setting Up a Spring Boot Batch Project
- Core Concepts of Spring Batch
- Configuring a Batch Job in Spring Batch
- Running the Batch Job
- JobLauncher in Detail
- Handling Job Restarts
- Handling Large Data with Chunk-Oriented Processing
- Parallel Processing in Spring Batch
What is Batch Processing?
Batch processing refers to the execution of a series of jobs or tasks without manual intervention. Unlike real-time processing, where tasks are executed immediately, batch processing handles large volumes of data in chunks or batches. These jobs are typically scheduled to run at specific times, such as overnight, when system usage is low.
Common use cases for batch processing include:
- Data Migration: Transferring data from one database to another.
- Report Generation: Aggregating data from various sources to create reports.
- Data Cleansing: Processing large datasets to identify and correct errors.
- ETL (Extract, Transform, Load) Processes: Extracting data from different sources, transforming it according to business rules, and loading it into a target database.
Spring Batch simplifies batch processing by providing reusable components, including readers, processors, and writers. Here are some of its key features:
Declarative I/O:
Spring Batch offers various ItemReader and ItemWriter implementations for reading from and writing to different data sources like files, databases, and message queues. This declarative approach abstracts the complexities of I/O operations.
Transaction Management:
Batch processing often involves interacting with databases and other transactional resources. Spring Batch provides built-in transaction management, ensuring data consistency and reliability.
Chunk-Oriented Processing:
Chunk-oriented processing is a core concept in Spring Batch. It breaks down large datasets into manageable chunks, processes them in memory, and commits them in a single transaction. This approach optimizes performance and resource usage.
Parallel Processing:
Spring Batch supports parallel processing, allowing jobs to be divided into smaller tasks that can run concurrently. This feature is essential for improving performance and reducing processing time in large-scale applications.
Job Monitoring and Restartability:
Spring Batch offers mechanisms to monitor job execution, handle failures, and restart jobs from the point of failure. This ensures that long-running jobs can be managed effectively in production environments.
Scalability:
Spring Batch is designed to handle large-scale batch processing jobs. Whether you’re processing millions of records or handling complex data transformations, Spring Batch provides the scalability required for enterprise-level applications.
Setting Up a Spring Boot Batch Project
To start with Spring Batch, you need to set up a Spring Boot project. You can use Spring Initializr or your preferred IDE to create a new Spring Boot project with the following dependencies:
- Spring Batch
- Spring Boot DevTools (Optional, for development convenience)
- Spring Data JPA (Optional, if using JPA for persistence)
- H2 Database (Optional, for in-memory database)
Here’s the pom.xml configuration:
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
</dependency>
</dependencies>Core Concepts of Spring Batch:
To understand Spring Batch, it’s essential to grasp its core concepts, which form the building blocks of batch processing in the framework:
Job:
A Job is the central concept in Spring Batch, representing the entire batch processing task. It encapsulates the sequence of steps and the flow of control needed to accomplish the batch process. Each job can contain one or more steps, and it defines the order and conditions under which these steps should be executed.
@Bean
public Job importUserJob(JobBuilderFactory jobBuilderFactory, Step step1, Step step2) {
return jobBuilderFactory.get("importUserJob")
.start(step1)
.next(step2)
.build();
}importUserJob is a job that starts with step1 and proceeds to step2.
Step:
A Step represents a single, independent phase of a job. It is the building block of a job and defines a specific task to be executed. Steps are reusable components that can be combined in different ways to form complex jobs.
Types of Steps:
- Tasklet Step: Executes a simple task using a
Tasklet. - Chunk-Oriented Step: Processes items in chunks using an
ItemReader,ItemProcessor, andItemWriter.
@Bean
public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<String> reader,
ItemProcessor<String, String> processor, ItemWriter<String> writer) {
return stepBuilderFactory.get("step1")
.<String, String>chunk(10)
.reader(reader)
.processor(processor)
.writer(writer)
.build();
}step1 is a chunk-oriented step that reads, processes, and writes items in chunks of 10.
Tasklet:
A Tasklet is a simple interface in Spring Batch that represents a single, atomic task within a step. It is often used for performing simple, one-off tasks that do not require chunk-oriented processing, such as file operations, sending notifications, or database cleanup.
@Bean
public Step taskletStep(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("taskletStep")
.tasklet(new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
System.out.println("Executing Tasklet Step");
return RepeatStatus.FINISHED;
}
}).build();
}taskletStep is a step that executes a simple tasklet, printing a message to the console.
ItemReader:
An ItemReader is responsible for reading data from a source, such as a file, database, or message queue, one item at a time. It acts as the input for the batch processing pipeline, feeding data into the ItemProcessor for processing.
@Bean
public FlatFileItemReader<UserClass> csvReader() {
return new FlatFileItemReaderBuilder<UserClass>()
.name("csvReader")
.resource(new ClassPathResource("data.csv"))
.delimited()
.names("field1", "field2", "field3")
.targetType(UserClass.class)
.build();
}In the above, csvReader is an ItemReader that reads data from a CSV file and maps it to an instance of UserClass .
ItemProcessor:
An ItemProcessor processes or transforms the data read by the ItemReader. It applies business logic, filtering, or transformations to the data before passing it to the ItemWriter for output.
@Bean
public ItemProcessor<UserClass, UserClass> csvProcessor() {
return item -> {
item.setField1(item.getField1().toUpperCase());
return item;
};
}Here, csvProcessor is an ItemProcessor that converts the field1 of YourClass to uppercase.
ItemWriter:
An ItemWriter is responsible for writing the processed data to an output destination, such as a file, database, or message queue. It acts as the final step in the batch processing pipeline.
@Bean
public ItemWriter<YourClass> csvWriter() {
return items -> {
for (YourClass item : items) {
System.out.println("Writing item: " + item);
}
};
}csvWriter is an ItemWriter that writes each item to the console.
Configuring a Batch Job in Spring Batch:
Spring Batch requires some infrastructure components such as a JobRepository, JobLauncher, and TransactionManager to function. These components manage job metadata, launch jobs, and handle transactions.
Create a BatchConfig.java file inside your src folder and add the following
@Configuration
@EnableBatchProcessing
public class BatchConfig {
private final JobBuilderFactory jobBuilderFactory;
private final StepBuilderFactory stepBuilderFactory;
public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
this.jobBuilderFactory = jobBuilderFactory;
this.stepBuilderFactory = stepBuilderFactory;
}
@Bean
public JobLauncher jobLauncher(JobRepository jobRepository) {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(jobRepository);
return jobLauncher;
}
@Bean
public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) {
JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
factory.setDataSource(dataSource);
factory.setTransactionManager(transactionManager);
factory.setDatabaseType(DatabaseType.MYSQL.name());
return factory.getObject();
}
@Bean
public PlatformTransactionManager transactionManager(DataSource dataSource) {
return new DataSourceTransactionManager(dataSource);
}
}JobRepository: Stores metadata about job executions, such as start/end times, status, and parameters.JobLauncher: Launches the batch job and manages its execution.TransactionManager: Manages transactions for steps, ensuring data consistency.
Defining a Job:
Now we can define a Job in a new java file as a Bean.
@Bean
public Job processJob(Step step1, Step step2) {
return jobBuilderFactory.get("processJob")
.start(step1)
.next(step2)
.build();
}This job processJob is composed of two steps (step1 and step2), executed sequentially. The steps are defined in a separate file like
@Bean
public Step taskletStep(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("taskletStep")
.tasklet((contribution, chunkContext) -> {
System.out.println("Tasklet executed");
return RepeatStatus.FINISHED;
}).build();
}Configuring an ItemReader, ItemProcessor and ItemWriter:
Spring Batch provides several out-of-the-box readers like FlatFileItemReader (for files) and JdbcCursorItemReader (for databases). We will configure one for .csv files
@Bean
public FlatFileItemReader<String> itemReader() {
return new FlatFileItemReaderBuilder<String>()
.name("csvReader")
.resource(new ClassPathResource("data.csv"))
.delimited()
.names(new String[]{"column1", "column2"})
.fieldSetMapper(new BeanWrapperFieldSetMapper<>() {
{
setTargetType(String.class);
}
}).build();
}
@Bean
public ItemProcessor<String, String> itemProcessor() {
return item -> {
// convert data to uppercase
return item.toUpperCase();
};
}
@Bean
public ItemWriter<String> itemWriter() {
return items -> {
for (String item : items) {
System.out.println("Writing item: " + item);
}
};
}Running the Batch Job:
Once the batch job and its components are configured, you can run it as part of a Spring Boot application. The Spring Batch infrastructure takes care of job execution, step processing, and transaction management.
@SpringBootApplication
public class BatchApplication {
public static void main(String[] args) {
SpringApplication.run(BatchApplication.class, args);
}
}This is all made possible by Inversion of Control and Dependency injection using the @Bean tag.
Configuring a Job Scheduler (Optional):
In many cases, batch jobs are scheduled to run at specific times or intervals. Spring Batch can integrate with Spring’s @Scheduled annotation to run jobs automatically.
@EnableScheduling
public class BatchScheduler {
private final JobLauncher jobLauncher;
private final Job processJob;
public BatchScheduler(JobLauncher jobLauncher, Job processJob) {
this.jobLauncher = jobLauncher;
this.processJob = processJob;
}
@Scheduled(cron = "0 0 12 * * ?") // Runs every day at noon
public void runJob() throws Exception {
JobParameters params = new JobParametersBuilder()
.addString("JobID", String.valueOf(System.currentTimeMillis()))
.toJobParameters();
jobLauncher.run(processJob, params);
}
}The JobLauncher In Detail:
The JobLauncher is an interface in Spring Batch that provides a way to start a Job. It takes a Job and a set of JobParameters and returns a JobExecution object, which contains details about the job execution status, start time, end time, and other metadata.
The core method in the JobLauncher interface is:
JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException,
JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException;- Job: The batch job to be executed.
- Job Parameters are essential for differentiating between job instances and passing dynamic data to jobs. Spring Batch requires that job parameters be unique for each job instance, otherwise, it may assume that the job has already been executed and will not run it again.
- JobExecution: The result of the job execution, containing information about the job’s status and metadata.
To use the JobLauncher, it must be configured as a Spring bean in your application context. In most cases, Spring Batch's SimpleJobLauncher implementation is used.
@Configuration
@EnableBatchProcessing
public class BatchConfig {
private final JobRepository jobRepository;
private final PlatformTransactionManager transactionManager;
public BatchConfig(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
this.transactionManager = transactionManager;
}
@Bean
public JobLauncher jobLauncher() throws Exception {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(jobRepository);
jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor()); // Optional: To run jobs asynchronously
jobLauncher.afterPropertiesSet();
return jobLauncher;
}
}Once configured, the JobLauncher can be used to start jobs programmatically. This can be done from a REST controller, a scheduled task, or any other Spring-managed component.
We saw the scheduled way already. Here is an example of starting a job through REST controller,
@RestController
public class JobController {
private final JobLauncher jobLauncher;
private final Job job;
public JobController(JobLauncher jobLauncher, Job job) {
this.jobLauncher = jobLauncher;
this.job = job;
}
@PostMapping("/run-job")
public ResponseEntity<String> runJob() {
try {
JobParameters jobParameters = new JobParametersBuilder()
.addString("JobID", String.valueOf(System.currentTimeMillis()))
.toJobParameters();
JobExecution jobExecution = jobLauncher.run(job, jobParameters);
return ResponseEntity.ok("Job executed successfully, status: " + jobExecution.getStatus());
} catch (Exception e) {
e.printStackTrace();
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Job execution failed: " + e.getMessage());
}
}
}After running a job, the JobExecution object contains valuable information about the job's status, including:
- Job Status: Indicates whether the job completed successfully, failed, or is still running.
- Exit Status: Provides more detailed information about the job’s termination condition.
- Job Instance ID: A unique identifier for the job instance.
- Start and End Time: Timestamps for when the job started and ended.
JobExecution jobExecution = jobLauncher.run(job, jobParameters);
System.out.println("Job Execution ID: " + jobExecution.getId());
System.out.println("Job Status: " + jobExecution.getStatus());
System.out.println("Job Exit Status: " + jobExecution.getExitStatus().getExitCode());
System.out.println("Job Start Time: " + jobExecution.getStartTime());
System.out.println("Job End Time: " + jobExecution.getEndTime());Handling Job Restarts
Spring Batch supports job restarts in cases where a job fails or stops unexpectedly. This is managed by the JobRepository, which tracks the state of each job execution. When a job is restarted, it resumes from the last successful step.
try {
JobExecution jobExecution = jobLauncher.run(job, jobParameters);
if (jobExecution.getStatus() == BatchStatus.FAILED) {
System.out.println("Job failed. Restarting...");
jobExecution = jobLauncher.run(job, jobParameters);
}
} catch (Exception e) {
e.printStackTrace();
}- BatchStatus.FAILED: Indicates that the job failed, triggering a restart.
- JobRepository: Keeps track of job execution progress and allows for resuming jobs.
Restartability Considerations:
Ensure that steps are idempotent, meaning they can be run multiple times without adverse effects and also use appropriate job parameters to control the scope of the restart.
Handling Large Data with Chunk-Oriented Processing:
Instead of processing all data at once, which can be memory-intensive and slow, chunk-oriented processing breaks the data into manageable chunks. Each chunk is read, processed, and written within a transaction, ensuring both efficiency and reliability.
This approach is particularly useful when dealing with large datasets, such as processing millions of records from a database or a large file, where loading all data into memory at once is impractical.
The workflow can be summarized as:
- Read: A chunk of items is read using an
ItemReader. - Process: Each item in the chunk is processed by an
ItemProcessor. - Write: The processed items are written out using an
ItemWriter.
If the chunk size is set to 10, for instance, 10 items will be read, processed, and written in a single transaction. If an error occurs during processing or writing, only the current chunk will be rolled back, not the entire job.
To configure a chunk-oriented step in Spring Batch, you define a Step that specifies the chunk size along with the ItemReader, ItemProcessor, and ItemWriter components.
@Configuration
@EnableBatchProcessing
public class BatchConfig {
@Bean
public Job job(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
return jobBuilderFactory.get("job")
.start(chunkStep(stepBuilderFactory))
.build();
}
@Bean
public Step chunkStep(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("chunkStep")
.<InputType, OutputType>chunk(10) // Define chunk size
.reader(itemReader())
.processor(itemProcessor())
.writer(itemWriter())
.build();
}
@Bean
public ItemReader<InputType> itemReader() {
return new CustomItemReader();
}
@Bean
public ItemProcessor<InputType, OutputType> itemProcessor() {
return new CustomItemProcessor();
}
@Bean
public ItemWriter<OutputType> itemWriter() {
return new CustomItemWriter();
}
}Parallel Processing in Spring Batch
Parallel processing in Spring Batch allows you to execute steps or chunks concurrently, significantly improving the performance of batch jobs that handle large datasets or require time-intensive operations. By distributing the workload across multiple threads or processes, you can reduce the overall execution time of your batch job.
Spring Batch offers several techniques for parallel processing:
- Multi-threaded Step
- Partitioning
- Remote Chunking
- Parallel Steps
Multi-threaded Step:
A multi-threaded step allows a single step to be executed by multiple threads concurrently. This approach is useful when the step’s processing logic is thread-safe, and the workload can be distributed across multiple threads.
@Bean
public Step multiThreadedStep(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("multiThreadedStep")
.<InputType, OutputType>chunk(10)
.reader(itemReader())
.processor(itemProcessor())
.writer(itemWriter())
.taskExecutor(taskExecutor()) // Enable multi-threading
.throttleLimit(4) // Limit the number of concurrent threads
.build();
}
@Bean
public TaskExecutor taskExecutor() {
return new SimpleAsyncTaskExecutor();
}The step is executed with up to 4 threads, processing chunks in parallel. This approach is effective when you want to increase throughput without changing the job’s structure.
Partitioning:
Partitioning divides the data into smaller partitions, each of which is processed by an independent thread or process. This technique is particularly useful when dealing with large datasets that can be logically divided into smaller subsets.
Key Components:
Partitioner: Determines how the data is divided into partitions.
Step: Each partition is processed by a separate step execution.
@Bean
public Step masterStep(StepBuilderFactory stepBuilderFactory, JobRepository jobRepository, PlatformTransactionManager transactionManager) {
return stepBuilderFactory.get("masterStep")
.partitioner("slaveStep", partitioner())
.step(slaveStep(stepBuilderFactory))
.gridSize(4) // Number of partitions
.taskExecutor(taskExecutor())
.build();
}
@Bean
public Partitioner partitioner() {
return new CustomPartitioner(); // Defines partitioning logic
}
@Bean
public Step slaveStep(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("slaveStep")
.<InputType, OutputType>chunk(10)
.reader(itemReader())
.processor(itemProcessor())
.writer(itemWriter())
.build();
}In this example, the data is split into 4 partitions, and each partition is processed by a separate slaveStep.
Remote Chunking:
Remote chunking distributes the reading and processing of data across multiple remote nodes while keeping the writing centralized. This technique is useful when you want to leverage the computing power of multiple machines.
Master Step: Reads data and sends it to remote workers.
Slave Step: Remote workers process the data.
Result Aggregation: Processed data is sent back to the master for writing.
Master Configuration:
@Bean
public Step masterStep(StepBuilderFactory stepBuilderFactory, JobRepository jobRepository, PlatformTransactionManager transactionManager) {
return stepBuilderFactory.get("masterStep")
.<InputType, OutputType>chunk(10)
.reader(itemReader())
.writer(masterItemWriter()) // Writes the final results
.taskExecutor(taskExecutor())
.build();
}Slave Configuration:
@Bean
public Step slaveStep(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("slaveStep")
.<InputType, OutputType>chunk(10)
.reader(slaveItemReader())
.processor(slaveItemProcessor())
.writer(slaveItemWriter())
.build();
}The master reads and sends data chunks to remote workers (slaves) for processing. Once processed, the data is sent back to the master for final writing.
Parallel Steps:
Parallel steps allow multiple steps within a job to be executed concurrently. Each step operates independently, and they can run simultaneously if there are no dependencies between them.
@Bean
public Job parallelJob(JobBuilderFactory jobBuilderFactory, Step step1, Step step2) {
return jobBuilderFactory.get("parallelJob")
.start(step1)
.split(taskExecutor())
.add(step2)
.build();
}
@Bean
public Step step1(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("step1")
.<InputType, OutputType>chunk(10)
.reader(itemReader1())
.processor(itemProcessor1())
.writer(itemWriter1())
.build();
}
@Bean
public Step step2(StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("step2")
.<InputType, OutputType>chunk(10)
.reader(itemReader2())
.processor(itemProcessor2())
.writer(itemWriter2())
.build();
}split(taskExecutor()) Specifies that step1 and step2 should run in parallel using the provided TaskExecutor. Each step can have its own reader, processor, and writer. This approach is ideal when you have multiple independent tasks that can be performed simultaneously, speeding up the overall job execution.
Conclusion:
In this tutorial, we’ve explored the essentials of Spring Batch and how to leverage it within a Spring Boot application to handle batch processing efficiently. Whether you’re working with large datasets, complex processing logic, or need reliable job execution, Spring Batch offers the tools and flexibility to meet your requirements.
This tutorial only provides a foundation to start building and optimizing batch processing solutions in your projects. There are plenty more about Spring Batch that you can read here.
I’ll never see them again. I know that. And they know that. And knowing this, we say farewell.
― Haruki Murakami, Kafka on the Shore
