Spring Batch With MySQL Example

In this example, we will build a simple Batch application using Spring Batch with MySQL in Java Spring Boot.

Spring Batch is an open source, lightweight framework which is designed for use in developing robust batch applications. Batch applications are usually required by systems that need to process large volume of data on daily basis.

Spring Batch is not designed for use as a scheduling framework. However, it can be used in combination with a scheduling framework such as Quartz, Control-M, etc.

Spring Boot provides spring-boot-starter-batch dependency.

This sample batch application that we will build in this example will do the following:

1. Read data from a MySQL database table.
2. Transform the data.
3. Write the final results to a MySQL database table.

Complete the following steps to build our sample batch application:

Create a Spring Boot Application

1. Go to Spring Initializr at https://start.spring.io and create a Spring Boot application with details as follows:
• Project: Choose Gradle Project or Maven Project.
• Language: Java
• Spring Boot: Latest stable version of Spring Boot is selected by default. So leave it as is.
• Project Metadata: Provide group name in the Group field. The group name is the id of the project. In Artifact field, provide the name of your project. In the package field, provide package name for your project. Next, select your preferred version of Java that is installed on your computer and is available on your local environment.
• Dependencies: Add dependencies for Spring Web, Spring Boot DevTools, Spring Batch, MySQL Driver, and Spring Data JPA.

Refer to the image below for example:


2. Click the GENERATE button and save/download the project zip bundle.
3. Extract the project to your preferred working directory.
4. Import the project in your preferred Java development IDE such as Eclipse or IntelliJ IDEA.

The following listing shows Spring Batch, MySQL Driver, and Spring Data JPA dependencies for Gradle and Maven project:

With Gradle

implementation 'org.springframework.boot:spring-boot-starter-batch'
implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
runtimeOnly 'mysql:mysql-connector-java'

With Maven

Add Application Configurations

Open the application.properties file and copy the code listed below. Do not forget to update the configuration values to make them relevant to your project:

# Server port
server.port = 8080

#mysql database connection
spring.datasource.url = jdbc:mysql://localhost:3306/my_test_db
spring.datasource.username = root
spring.datasource.password = Testing123
spring.datasource.timeBetweenEvictionRunsMillis = 60000
spring.datasource.maxIdle = 1
spring.jpa.generate-ddl=true                                                        
spring.jpa.hibernate.ddl-auto=update
spring.batch.initialize-schema=ALWAYS

#disabled job run at startup
spring.batch.job.enabled=false

Create Entity classes

Create two classes, User.java class to map with the user table and Profile.java class to map with the profile table of MySQL database. We will read data from from the user table and add the data in the profile table. Annotate these classes with @Entity, @Table(name = "table_name"), @EntityListeners(AuditingEntityListener.class). The @entity annotation identifies a class as being a entity object that we want to persist to the database. The attributes of the entity class are mapped to the columns of the database table. The following code shows how to do so:

package com.example.domain;

import java.time.LocalDateTime;
import javax.persistence.Entity;
import javax.persistence.EntityListeners;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.Table;
import org.springframework.data.annotation.CreatedDate;
import org.springframework.data.annotation.LastModifiedDate;
import org.springframework.data.jpa.domain.support.AuditingEntityListener;

@Entity
@Table(name = "user")
@EntityListeners(AuditingEntityListener.class)
public class User {
    @Id
    @GeneratedValue(strategy = GenerationType.AUTO)
    private Integer id;
    private String email;
    private boolean emailVerified;
    private String firstName;
    private String lastName;
    private String status;
    @CreatedDate
    private LocalDateTime createdDate;
    @LastModifiedDate
    private LocalDateTime lastModifiedDate;
    public Integer getId() {
        return id;
    }
    public void setId(Integer id) {
        this.id = id;
    }
    public String getEmail() {
        return email;
    }
    public void setEmail(String email) {
        this.email = email;
    }
    public boolean isEmailVerified() {
        return emailVerified;
    }
    public void setEmailVerified(boolean emailVerified) {
        this.emailVerified = emailVerified;
    }
    public String getFirstName() {
        return firstName;
    }
    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }
    public String getLastName() {
        return lastName;
    }
    public void setLastName(String lastName) {
        this.lastName = lastName;
    }
    public String getStatus() {
        return status;
    }
    public void setStatus(String status) {
        this.status = status;
    }
    public LocalDateTime getCreatedDate() {
        return createdDate;
    }
    public void setCreatedDate(LocalDateTime createdDate) {
        this.createdDate = createdDate;
    }
    public LocalDateTime getLastModifiedDate() {
        return lastModifiedDate;
    }
    public void setLastModifiedDate(LocalDateTime lastModifiedDate) {
        this.lastModifiedDate = lastModifiedDate;
    }
        
}

package com.example.domain;

import java.time.LocalDateTime;
import javax.persistence.Entity;
import javax.persistence.EntityListeners;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.Table;
import org.springframework.data.annotation.CreatedDate;
import org.springframework.data.annotation.LastModifiedDate;
import org.springframework.data.jpa.domain.support.AuditingEntityListener;

@Entity
@Table(name = "profile")
@EntityListeners(AuditingEntityListener.class)
public class Profile {

    @Id
    @GeneratedValue(strategy = GenerationType.AUTO)
    private Integer id;
    private String fullName;
    private Integer userId;
    @CreatedDate
    private LocalDateTime createdDate;
    @LastModifiedDate
    private LocalDateTime lastModifieDate;

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public String getFullName() {
        return fullName;
    }

    public void setFullName(String fullName) {
        this.fullName = fullName;
    }

    public Integer getUserId() {
        return userId;
    }

    public void setUserId(Integer userId) {
        this.userId = userId;
    }

    public LocalDateTime getCreatedDate() {
        return createdDate;
    }

    public void setCreatedDate(LocalDateTime createdDate) {
        this.createdDate = createdDate;
    }

    public LocalDateTime getLastModifieDate() {
        return lastModifieDate;
    }

    public void setLastModifieDate(LocalDateTime lastModifieDate) {
        this.lastModifieDate = lastModifieDate;
    }

}

Create Repository Interfaces

Create two interfaces, UserRepository.java interface to perform CRUD operations on the user table and ProfileRepository.java interface to perform CRUD operations on the profile table of MySQL database. The repository interface must be annotated with @Repository annotation. The following code shows how to do so:

package com.example.repository;

import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.repository.PagingAndSortingRepository;
import org.springframework.stereotype.Repository;
import com.example.domain.User;


@Repository
public interface UserRepository extends PagingAndSortingRepository<User, Integer> {

    Page<User> findByStatusAndEmailVerified(String status, boolean emailVerified,
            Pageable pageable);
}

package com.example.repository;

import org.springframework.data.repository.CrudRepository;
import org.springframework.stereotype.Repository;
import com.example.domain.Profile;

@Repository
public interface ProfileRepository extends CrudRepository<Profile, Integer>{

}

Create a Processor class

The work of a processor class is to transform data. Create a ProfileItemProcessor.java class and implement the ItemProcessor<I, O> interface of the Spring Batch framework.

The ItemProcessor<I, O> interface takes two arguments, Input type and Output type. Both doesn't need to be of the same type. You may provide input of one type and return output of some other type after it has been read.

The ItemProcessor<I, O> has a public method that takes an argument of object with data that is read. This method is where the read data must be transformed.

package com.example.processor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import com.example.domain.Profile;
import com.example.domain.User;



public class ProfileItemProcessor implements ItemProcessor<User, Profile> {
    private static final Logger log = LoggerFactory.getLogger(ProfileItemProcessor.class);

    @Override
    public Profile process(User user) throws Exception {

        log.info("processing user data.....{}", user);

        Profile transformedProfile = new Profile();
        transformedProfile.setUserId(user.getId());
        transformedProfile.setFullName(user.getFirstName() + " " + user.getLastName());
        return transformedProfile;
    }

}

Create a Job Notification Listener class

Create a JobCompletionNotificationListener.java Java class and extend the class with JobExecutionListenerSupport class from the Spring Batch framework. The JobExecutionListenerSupport class has callback methods which can be called before starting or after completion of a job.

Let's override the afterJob method of JobExecutionListenerSupport class which will help the application to notify when the job completes. The following code shows how to do so:

package com.example.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
import org.springframework.stereotype.Component;
@ Component
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
  
    private static final Logger log =
            LoggerFactory.getLogger(JobCompletionNotificationListener.class);

    @Override
    public void afterJob(JobExecution jobExecution) {
        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
            log.info("!!! JOB FINISHED! Time to verify the results");

        }
    }
}

Create a BatchConfiguration class

Create a BatchConfiguration.java class. In this class, we will do all the batch configurations. Annotate the class with @Configuration and @EnableBatchProcessing annotations. The @EnableBatchProcessing enables Spring Batch features and provide a base configuration for setting up batch jobs in a @Configuration class. The following code shows how to do so:

package com.example.config;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.data.RepositoryItemReader;
import org.springframework.batch.item.data.RepositoryItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.domain.Sort.Direction;
import com.example.domain.Profile;
import com.example.domain.User;
import com.example.listener.JobCompletionNotificationListener;
import com.example.processor.ProfileItemProcessor;
import com.example.repository.ProfileRepository;
import com.example.repository.UserRepository;


@Configuration
@EnableBatchProcessing
public class BatchConfiguration {
    @Autowired
    public JobBuilderFactory jobBuilderFactory;
    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    @Autowired
    @Lazy
    private UserRepository userRepository;

    @Autowired
    @Lazy
    private ProfileRepository profileRepository;


    @Bean
    public RepositoryItemReader<User> reader() {
        RepositoryItemReader<User> reader = new RepositoryItemReader<>();
        reader.setRepository(userRepository);
        reader.setMethodName("findByStatusAndEmailVerified");

        List<Object> queryMethodArguments = new ArrayList<>();
        // for status
        queryMethodArguments.add("APPROVED");
        // for emailVerified
        queryMethodArguments.add(Boolean.TRUE);

        reader.setArguments(queryMethodArguments);
        reader.setPageSize(100);
        Map<String, Direction> sorts = new HashMap<>();
        sorts.put("id", Direction.ASC);
        reader.setSort(sorts);

        return reader;
    }

    @Bean
    public RepositoryItemWriter<Profile> writer() {
        RepositoryItemWriter<Profile> writer = new RepositoryItemWriter<>();
        writer.setRepository(profileRepository);
        writer.setMethodName("save");
        return writer;
    }


    @Bean
    public ProfileItemProcessor processor() {
        return new ProfileItemProcessor();
    }


    @Bean
    public Step step1(ItemReader<User> itemReader, ItemWriter<Profile> itemWriter)
            throws Exception {

        return this.stepBuilderFactory.get("step1").<User, Profile>chunk(5).reader(itemReader)
                .processor(processor()).writer(itemWriter).build();
    }

    @Bean
    public Job profileUpdateJob(JobCompletionNotificationListener listener, Step step1)
            throws Exception {

        return this.jobBuilderFactory.get("profileUpdateJob").incrementer(new RunIdIncrementer())
                .listener(listener).start(step1).build();
    }

}

Create a Web Controller class

Create a BatchController.java Java class with a POST method to run the batch. The following code shows how to do so:

package com.example.controller;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(path = "/batch")
public class BatchController {
    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private Job job;

    @PostMapping(path = "/start")
    public void startBatch() {
        JobParameters jobParameters = new JobParametersBuilder()
                .addLong("startAt", System.currentTimeMillis()).toJobParameters();
        try {
            jobLauncher.run(job, jobParameters);
        } catch (JobExecutionAlreadyRunningException | JobRestartException
                | JobInstanceAlreadyCompleteException | JobParametersInvalidException e) {
            
            e.printStackTrace();
        }
    }
}

Update Main Application class

Annotate the application main class with @EnableJpaAuditing annotation. The following code shows to do so:

package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.data.jpa.repository.config.EnableJpaAuditing;

@EnableJpaAuditing
@SpringBootApplication
public class BatchMysqlSampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(BatchMysqlSampleApplication.class, args);
    }

}

Run the Application

Run your batch application and make a POST request to /batch/start path. The application should read the CSV file and save the data to the MongoDB database.

Summary

Congratulations! you have learned how to build a Spring Batch application with MySQL and Spring Boot.