Spring Batch provides a FlatFileItemReader that we can use to read data from flat files, including CSV files. Here’s an example of how to configure and use FlatFileItemReader to read data from a CSV file in a Spring Batch job.
1. CSV File and Model
For demo purposes, we will be using the following CSF files:
Lokesh,Gupta,41,true
Brian,Schultz,42,false
John,Cena,43,true
Albert,Pinto,44,falseThen we need to create a domain object to represent the data.
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Person {
String firstName;
String lastName;
Integer age;
Boolean active;
}2. Configuring FlatFileItemReader
The org.springframework.batch.item.file.FlatFileItemReader consists of two main components:
- A Spring Resource that represents the file to be read
- An implementation of the LineMapper interface (same as RowMapper in Spring JDBC). When reading a flat file, each line is presented to LineMapper as a String to parse.
The LineMapper internally consists of a LineTokenizer and FieldSetMapper. The LineTokenizer implementation parses the line into a FieldSet (similar to columns in a database row). The FieldSetMapper later maps the FieldSets to a domain object.
2.1. Delimited Files (CSV Files)
In delimited files, a character acts as a divider between each field in the record. In delimited files, we map the columns to the POJO fields after dividing each record with the delimiter. The default delimiter is always a comma.
A Step configuration that can read the delimited flat file can be built using the FlatFileItemReaderBuilder.
@Bean
@StepScope
public FlatFileItemReader<Person> personItemReader() {
return new FlatFileItemReaderBuilder<Person>()
.name("personItemReader")
.delimited()
.names("firstName", "lastName", "age", "active")
.targetType(Person.class)
.resource(csvFile)
.build();
}If we want to configure a different delimiter, we can define the custom DelimitedLineTokenizer bean.
@Bean
public DelimitedLineTokenizer tokenizer() {
var tokenizer = new DelimitedLineTokenizer();
tokenizer.setDelimiter("#"); // Specify a different delimiter. Default is comma.
tokenizer.setNames("firstName", "lastName", "age", "active");
return tokenizer;
}2.2. Fixed-Width Files
When working on legacy mainframe systems, we may encounter fixed-width files due to the way COBOL and other such technologies declare their storage.
In the absence of a delimiter (or any other metadata), we have to rely on the length of each field in the file. Consider the following fixed-width file:
Lokesh Gupta 41 true
Brian Schultz 42 false
John Cena 43 true
Albert Pinto 44 falseIn the above file, the lengths of the fields are:
| firstName | 10 |
| lastName | 10 |
| age | 4 |
| active | 5 |
The equivalent FlatFileItemReader can be used by using the methods .fixedLength() and columns() specifying the length of the fields.
@Bean
@StepScope
public FlatFileItemReader<Person> personItemReaderFixedWidth() {
return new FlatFileItemReaderBuilder<Person>()
.name("personItemReader")
.fixedLength()
.columns(new Range(1, 10), new Range(11, 20), new Range(21, 24), new Range(25, 30))
.names("firstName", "lastName", "age", "active")
.targetType(Person.class)
.resource(csvFile)
.build();
}2.3. FieldSetMapper
By default, Spring batch uses BeanWrapperFieldSetMapper which is a FieldSetMapper implementation based on a fuzzy search of bean property paths. It makes a good guess to match the column names with the field names in the POJO class. For example, the BeanWrapperFieldSetMapper will call Person#setFirstName, Person#setLastName, and so on, based on the names of the columns configured in the LineTokenizer.
If there is quite a difference in the column manes and the PJO class field names or structure of fields, we can define our own implementation of FieldSetMapper.
public class PersonFieldSetMapper implements FieldSetMapper<Person> {
public Person mapFieldSet(FieldSet fieldSet) {
Person person = new Person();
person.setFirstName(fieldSet.readString("firstName"));
person.setLastName(fieldSet.readString("lastName"));
....
return person;
}
}And then inject this PersonFieldSetMapper into FlatFileItemReaderBuilder as follows:
@Bean
@StepScope
public FlatFileItemReader<Person> personItemReader() {
return new FlatFileItemReaderBuilder<Person>()
.name("personItemReader")
.delimited()
.names("firstName", "lastName", "age", "active")
.fieldSetMapper(new PersonFieldSetMapper())
.resource(csvFile)
.build();
}3. Read CSV with FlatFileItemReader
In the following configuration, the FlatFileItemReader is configured to read a CSV file. The DelimitedLineTokenizer is used to specify the column names, and the BeanWrapperFieldSetMapper is used to map each line to a Person object.
We’ll need to customize the ItemProcessor and ItemWriter beans according to the business logic and data destination. This configuration writes data to the database.
Finally, create a Job that includes the Steps.
import com.howtodoinjava.demo.batch.jobs.csvToDb.model.Person;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.LineMapper;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.LineTokenizer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.transaction.PlatformTransactionManager;
import javax.sql.DataSource;
@Configuration
public class CsvToDatabaseJob {
public static final Logger logger = LoggerFactory.getLogger(CsvToDatabaseJob.class);
private static final String INSERT_QUERY = """
insert into person (first_name, last_name, age, is_active)
values (:firstName,:lastName,:age,:active)""";
private final JobRepository jobRepository;
public CsvToDatabaseJob(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
@Value("classpath:csv/inputData.csv")
private Resource inputFeed;
@Bean(name="insertIntoDbFromCsvJob")
public Job insertIntoDbFromCsvJob(Step step1, Step step2) {
var name = "Persons Import Job";
var builder = new JobBuilder(name, jobRepository);
return builder.start(step1).build();
}
@Bean
public Step step1(ItemReader<Person> reader,
ItemWriter<Person> writer,
ItemProcessor<Person, Person> processor,
PlatformTransactionManager txManager) {
var name = "INSERT CSV RECORDS To DB Step";
var builder = new StepBuilder(name, jobRepository);
return builder
.reader(reader)
.writer(writer)
.build();
}
@Bean
public FlatFileItemReader<Person> reader(
LineMapper<Person> lineMapper) {
var itemReader = new FlatFileItemReader<Person>();
itemReader.setLineMapper(lineMapper);
itemReader.setResource(inputFeed);
return itemReader;
}
@Bean
public DefaultLineMapper<Person> lineMapper(LineTokenizer tokenizer,
FieldSetMapper<Person> fieldSetMapper) {
var lineMapper = new DefaultLineMapper<Person>();
lineMapper.setLineTokenizer(tokenizer);
lineMapper.setFieldSetMapper(fieldSetMapper);
return lineMapper;
}
@Bean
public BeanWrapperFieldSetMapper<Person> fieldSetMapper() {
var fieldSetMapper = new BeanWrapperFieldSetMapper<Person>();
fieldSetMapper.setTargetType(Person.class);
return fieldSetMapper;
}
@Bean
public DelimitedLineTokenizer tokenizer() {
var tokenizer = new DelimitedLineTokenizer();
tokenizer.setDelimiter(",");
tokenizer.setNames("firstName", "lastName", "age", "active");
return tokenizer;
}
@Bean
public JdbcBatchItemWriter<Person> writer(DataSource dataSource) {
var provider = new BeanPropertyItemSqlParameterSourceProvider<Person>();
var itemWriter = new JdbcBatchItemWriter<Person>();
itemWriter.setDataSource(dataSource);
itemWriter.setSql(INSERT_QUERY);
itemWriter.setItemSqlParameterSourceProvider(provider);
return itemWriter;
}
}If the above configuration seems like a lot then you can merge the DefaultLineMapper, DelimitedLineTokenizer and BeanWrapperFieldSetMapper in the FlatFileItemReader bean itself.
@Bean
public FlatFileItemReader<Person> reader() {
FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
reader.setResource(inputFile);
reader.setLineMapper(new DefaultLineMapper<Person>() {{
setLineTokenizer(new DelimitedLineTokenizer() {{
setNames("firstName", "lastName", "age", "active");
}});
setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
setTargetType(Person.class);
}});
}});
return reader;
} 4. Demo
4.1. Maven
Make sure you have the following dependencies in the project:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>runtime</scope>
</dependency>4.2. Run the Application
Now run the application, and watch out for the console logs.
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.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;
@SpringBootApplication
public class BatchProcessingApplication implements CommandLineRunner {
private final JobLauncher jobLauncher;
private final ApplicationContext applicationContext;
public BatchProcessingApplication(JobLauncher jobLauncher, ApplicationContext applicationContext) {
this.jobLauncher = jobLauncher;
this.applicationContext = applicationContext;
}
public static void main(String[] args) {
SpringApplication.run(BatchProcessingApplication.class, args);
}
@Override
public void run(String... args) throws Exception {
Job job = (Job) applicationContext.getBean("insertIntoDbFromCsvJob");
JobParameters jobParameters = new JobParametersBuilder()
.addString("JobID", String.valueOf(System.currentTimeMillis()))
.toJobParameters();
var jobExecution = jobLauncher.run(job, jobParameters);
var batchStatus = jobExecution.getStatus();
while (batchStatus.isRunning()) {
System.out.println("Still running...");
Thread.sleep(5000L);
}
}
}
The program output:
2023-11-29T14:32:54.612+05:30 INFO 24044 --- [main] o.s.b.c.l.support.SimpleJobLauncher : Job: [SimpleJob: [name=Persons Import Job]] launched with the following parameters: [{'JobID':'{value=1701248574579, type=class java.lang.String, identifying=true}'}]
2023-11-29T14:32:54.631+05:30 INFO 24044 --- [main] o.s.batch.core.job.SimpleStepHandler : Executing step: [INSERT CSV RECORDS To DB Step]
2023-11-29T14:32:54.647+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : Reading a new Person Record
2023-11-29T14:32:54.662+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : New Person record read : Person(firstName=Lokesh, lastName=Gupta, age=41, active=true)
2023-11-29T14:32:54.664+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : Reading a new Person Record
2023-11-29T14:32:54.665+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : New Person record read : Person(firstName=Brian, lastName=Schultz, age=42, active=false)
2023-11-29T14:32:54.665+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : Reading a new Person Record
2023-11-29T14:32:54.665+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : New Person record read : Person(firstName=John, lastName=Cena, age=43, active=true)
2023-11-29T14:32:54.666+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : Reading a new Person Record
2023-11-29T14:32:54.666+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : New Person record read : Person(firstName=Albert, lastName=Pinto, age=44, active=false)
2023-11-29T14:32:54.666+05:30 INFO 24044 --- [main] c.h.d.b.j.c.l.PersonItemReadListener : Reading a new Person Record
2023-11-29T14:32:54.676+05:30 INFO 24044 --- [main] o.s.batch.core.step.AbstractStep : Step: [INSERT CSV RECORDS To DB Step] executed in 44ms
2023-11-29T14:32:54.679+05:30 INFO 24044 --- [main] .j.c.l.JobCompletionNotificationListener : JOB FINISHED !!
Drop me your questions in the comments section.
Happy Learning !!
Comments