– Tutorial “Spring Boot and Spring Batch Integration Example – Import CSV data to PostgreSQL database using Spring Batch Job”
Spring Batch is a powerful module to implement a batch process for tons of data conveniently. This tutorial guide you how to import CSV Data to PostgreSQL Database using Spring Batch Job.
* Technologies:
– Java 1.8
– Maven
– Spring Tool Suite
– Spring Boot
Project Goal – Spring Boot and Spring Batch Integration Example
To build a simple application that expresses how we can use Spring Batch with Job Step (including ItemReader, ItemProcessor, ItemWriter and JobExecutionListener) to read Customer Data from CSV file, then put them to PostgreSQL Table named ‘customer’.
– Reader: reads content of CSV file, then maps the read data to fields of DataModel Customer.
– Processor: converts each Customer record’s content to new content (for example, get Random ID and uppercase Name String) which will be written to Database Table.
– Writer: writes batch of records to PostgreSQL Database using DAO.
– Listener: handles after Step, read data from Database Table to verify and show LOGs.
– RestController: runs Job using JobLauncher, then return Complete String to Client.
– DAO: interacts with Database.
* Project Structure:
Step by step build Spring Boot and Spring Batch Integration Example
– Create Spring Boot project
– Configure application properties
– Create a DataModel
– Create a DAO
– Create Job Step: Reader, Processor, Writer, Listener
– Create Batch Configuration
– Create a WebController
– Run Spring Boot Application & Enjoy Result
Create Spring Boot project – Spring Boot and Spring Batch Integration Example
– Open Spring Tool Suite, on Menu, choose File -> New -> Spring Starter Project, then fill each fields:
Click Next, in:
– I/O: choose Batch
– SQL: choose JDBC and PostgreSQL
– Web: choose Web
Click Finish. Spring Boot project will be created successfully.
Configure application properties, add SQL Script File & CSV file
– Add configuration for Datasource and disable Spring Batch Job auto-run in application.properties (locates in src/main/resources):
spring.datasource.url=jdbc:postgresql://localhost:5432/testcsvdb
spring.datasource.username=postgres
spring.datasource.password=123
spring.datasource.platform=postgresql
spring.batch.job.enabled=false
– under src/main/resources:
+ schema-postgresql.sql
DROP TABLE IF EXISTS customer;
CREATE TABLE customer (
id Bigserial PRIMARY KEY NOT NULL,
first_name VARCHAR(20),
last_name VARCHAR(20)
);
+ customer-data.csv
0,Jack,Smith
1,Adam,Johnson
2,Kim,Smith
3,David,Williams
4,Peter,Davis
Create a SpringBoot DataModel – Spring Boot and Spring Batch Integration Example
Under package model, create class Customer.
Content of Customer.java:
package com.loizenai.batchcsvpostgresql.model;
public class Customer {
private long id;
private String firstName;
private String lastName;
public Customer() {
}
public Customer(long id, String firstName, String lastName) {
this.id = id;
this.firstName = firstName;
this.lastName = lastName;
}
public long getId() {
return id;
}
public void setId(long id) {
this.id = id;
}
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;
}
@Override
public String toString() {
return String.format("Customer[id=%d , firstName='%s', lastName='%s']", id, firstName, lastName);
}
}Create a SpringBoot DAO – Spring Boot and Spring Batch Integration Example
– Under package dao, create interface CustomerDao:
package com.loizenai.batchcsvpostgresql.dao;
import java.util.List;
import com.loizenai.batchcsvpostgresql.model.Customer;
public interface CustomerDao {
public void insert(List customers);
List loadAllCustomers();
}
– Under package dao.impl, create implementation of CustomerDao – CustomerDaoImpl:
package com.loizenai.batchcsvpostgresql.dao.impl;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import javax.annotation.PostConstruct;
import javax.sql.DataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.support.JdbcDaoSupport;
import org.springframework.stereotype.Repository;
import com.loizenai.batchcsvpostgresql.dao.CustomerDao;
import com.loizenai.batchcsvpostgresql.model.Customer;
@Repository
public class CustomerDaoImpl extends JdbcDaoSupport implements CustomerDao {
@Autowired
DataSource dataSource;
@PostConstruct
private void initialize() {
setDataSource(dataSource);
}
@Override
public void insert(List Customers) {
String sql = "INSERT INTO customer " + "(id, first_name, last_name) VALUES (?, ?, ?)";
getJdbcTemplate().batchUpdate(sql, new BatchPreparedStatementSetter() {
public void setValues(PreparedStatement ps, int i) throws SQLException {
Customer customer = Customers.get(i);
ps.setLong(1, customer.getId());
ps.setString(2, customer.getFirstName());
ps.setString(3, customer.getLastName());
}
public int getBatchSize() {
return Customers.size();
}
});
}
@Override
public List loadAllCustomers() {
String sql = "SELECT * FROM customer";
List> rows = getJdbcTemplate().queryForList(sql);
List result = new ArrayList();
for (Map row : rows) {
Customer customer = new Customer();
customer.setId((Long) row.get("id"));
customer.setFirstName((String) row.get("first_name"));
customer.setLastName((String) row.get("last_name"));
result.add(customer);
}
return result;
}
} Create Job Step: Reader, Processor, Writer, Listener
In the tutorial “Spring Boot and Spring Batch Integration Example”, under package step:
We Reader.java:
package com.loizenai.batchcsvpostgresql.step;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.core.io.ClassPathResource;
import com.loizenai.batchcsvpostgresql.model.Customer;
public class Reader {
public static FlatFileItemReader reader(String path) {
FlatFileItemReader reader = new FlatFileItemReader();
reader.setResource(new ClassPathResource(path));
reader.setLineMapper(new DefaultLineMapper() {
{
setLineTokenizer(new DelimitedLineTokenizer() {
{
setNames(new String[] { "id", "firstName", "lastName" });
}
});
setFieldSetMapper(new BeanWrapperFieldSetMapper() {
{
setTargetType(Customer.class);
}
});
}
});
return reader;
}
} Processor.java:
package com.loizenai.batchcsvpostgresql.step;
import java.util.Random;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import com.loizenai.batchcsvpostgresql.model.Customer;
public class Processor implements ItemProcessor {
private static final Logger log = LoggerFactory.getLogger(Processor.class);
@Override
public Customer process(Customer customer) throws Exception {
Random r = new Random();
final String firstName = customer.getFirstName().toUpperCase();
final String lastName = customer.getLastName().toUpperCase();
final Customer fixedCustomer = new Customer(r.nextLong(), firstName, lastName);
log.info("Converting (" + customer + ") into (" + fixedCustomer + ")");
return fixedCustomer;
}
}
Writer.java:
package com.loizenai.batchcsvpostgresql.step;
import java.util.List;
import org.springframework.batch.item.ItemWriter;
import com.loizenai.batchcsvpostgresql.dao.CustomerDao;
import com.loizenai.batchcsvpostgresql.model.Customer;
public class Writer implements ItemWriter {
private final CustomerDao customerDao;
public Writer(CustomerDao customerDao) {
this.customerDao = customerDao;
}
@Override
public void write(List customers) throws Exception {
customerDao.insert(customers);
}
} Listener.java:
package com.loizenai.batchcsvpostgresql.step;
import java.util.List;
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 com.loizenai.batchcsvpostgresql.dao.CustomerDao;
import com.loizenai.batchcsvpostgresql.model.Customer;
public class Listener extends JobExecutionListenerSupport {
private static final Logger log = LoggerFactory.getLogger(Listener.class);
private final CustomerDao customerDao;
public Listener(CustomerDao customerDao) {
this.customerDao = customerDao;
}
@Override
public void afterJob(JobExecution jobExecution) {
if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
log.info("Finish Job! Check the results");
List customers = customerDao.loadAllCustomers();
for (Customer customer : customers) {
log.info("Found <" + customer + "> in the database.");
}
}
}
}
JobExecutionListenerSupport is an implementation of JobExecutionListener.
We can make our own operations before start of a Step (override beforeJob method) and after its ending (normally or failed by overriding afterJob method).
The annotations corresponding to this interface are @BeforeStep and @AfterStep.
Create Batch Configuration
Under package config, create BatchConfig.java:
package com.loizenai.batchcsvpostgresql.config;
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.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.loizenai.batchcsvpostgresql.dao.CustomerDao;
import com.loizenai.batchcsvpostgresql.model.Customer;
import com.loizenai.batchcsvpostgresql.step.Listener;
import com.loizenai.batchcsvpostgresql.step.Processor;
import com.loizenai.batchcsvpostgresql.step.Reader;
import com.loizenai.batchcsvpostgresql.step.Writer;
@Configuration
@EnableBatchProcessing
public class BatchConfig {
@Autowired
public JobBuilderFactory jobBuilderFactory;
@Autowired
public StepBuilderFactory stepBuilderFactory;
@Autowired
public CustomerDao customerDao;
@Bean
public Job job() {
return jobBuilderFactory.get("job").incrementer(new RunIdIncrementer()).listener(new Listener(customerDao))
.flow(step1()).end().build();
}
@Bean
public Step step1() {
return stepBuilderFactory.get("step1").chunk(2)
.reader(Reader.reader("customer-data.csv"))
.processor(new Processor()).writer(new Writer(customerDao)).build();
}
} The input parameter for method chunk of StepBuilder specifies the number of items to read before writing out via the ItemWriter.
Create a SpringBoot RestAPI WebController
Under package controller, create WebController.java:
package com.loizenai.batchcsvpostgresql.controller;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class WebController {
@Autowired
JobLauncher jobLauncher;
@Autowired
Job job;
@RequestMapping("/runjob")
public String handle() throws Exception {
Logger logger = LoggerFactory.getLogger(this.getClass());
try {
JobParameters jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis())
.toJobParameters();
jobLauncher.run(job, jobParameters);
} catch (Exception e) {
logger.info(e.getMessage());
}
return "Done! Check Console Window for more details";
}
}
Run Spring Boot Application & Enjoy Result
– Config maven build:
clean install
– Run project with mode Spring Boot App
– Check results by access:
http://localhost:8080/runjob
System shows:
Job: [FlowJob: [name=job]] launched with the following parameters: [{time=1474771650234}]
Executing step: [step1]
Converting (Customer[id=0 , firstName='Jack', lastName='Smith']) into (Customer[id=6323287188840625065 , firstName='JACK', lastName='SMITH'])
Converting (Customer[id=1 , firstName='Adam', lastName='Johnson']) into (Customer[id=-5061932588533513687 , firstName='ADAM', lastName='JOHNSON'])
Converting (Customer[id=2 , firstName='Kim', lastName='Smith']) into (Customer[id=250312719692371085 , firstName='KIM', lastName='SMITH'])
Converting (Customer[id=3 , firstName='David', lastName='Williams']) into (Customer[id=2361081371024882848 , firstName='DAVID', lastName='WILLIAMS'])
Converting (Customer[id=4 , firstName='Peter', lastName='Davis']) into (Customer[id=-8239802131547868532 , firstName='PETER', lastName='DAVIS'])
Finish Job! Check the results
Found in the database.
Found in the database.
Found in the database.
Found in the database.
Found in the database.
Job: [FlowJob: [name=job]] completed with the following parameters: [{time=1474771650234}] and the following status: [COMPLETED]
Open phAdminIII to check table customer in database testcsvdb:
