jobconfig.java
package com.mugil.org.config;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.mugil.org.model.Employee;
import com.mugil.org.model.EmployeeCSV;
import com.mugil.org.model.EmployeeJSON;
import com.mugil.org.model.EmployeeJdbc;
import org.springframework.batch.core.*;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
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.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.LineMapper;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor;
import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.LineAggregator;
import org.springframework.batch.item.json.JacksonJsonObjectReader;
import org.springframework.batch.item.json.JsonItemReader;
import org.springframework.batch.item.json.builder.JsonItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.transaction.PlatformTransactionManager;
import javax.sql.DataSource;
import java.util.Date;
@Configuration
public class JobConfig {
@Autowired
ItemProcessor simpleJobProcessor;
@Autowired
DataSource dataSource;
@Bean
public Job job(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
return new JobBuilder("job", jobRepository)
.incrementer(new RunIdIncrementer())
.start(simpleChunkStep(jobRepository, transactionManager))
.build();
}
@Bean
public Step simpleChunkStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
StepBuilder stepBuilderOne = new StepBuilder("Chunk Oriented Step", jobRepository);
return stepBuilderOne
.chunk(3, transactionManager)
.reader(flatFileItemReader())
.processor(simpleJobProcessor)
.writer(jdbcWriter())
.build();
}
@Bean
public JsonItemReader<EmployeeJSON> jsonFileReader() {
return new JsonItemReaderBuilder<EmployeeJSON>()
.name("employeeJSONReader")
.jsonObjectReader(new JacksonJsonObjectReader<>(EmployeeJSON.class))
.currentItemCount(1) //Skip the first item
.maxItemCount(3) //Limit reading to a maximum of 3 items from the file
.resource(new FileSystemResource("EmployeeList.json"))
.build();
}
@Bean
public FlatFileItemWriter<EmployeeJSON> jsonFileWriter() {
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.enable(SerializationFeature.INDENT_OUTPUT); // Optional: pretty print
LineAggregator<EmployeeJSON> jsonLineAggregator = item -> {
try {
return objectMapper.writeValueAsString(item);
} catch (JsonProcessingException e) {
throw new RuntimeException("Error converting Employee to JSON", e);
}
};
return new FlatFileItemWriterBuilder<EmployeeJSON>()
.name("employeeJsonTxtWriter")
.resource(new FileSystemResource("OutputFiles/employees.json"))
.lineAggregator(jsonLineAggregator)
.build();
}
@Bean
public FlatFileItemReader<Employee> flatFileItemReader() {
FlatFileItemReader<Employee> flatFileItemReader = new FlatFileItemReader<>();
flatFileItemReader.setResource(new FileSystemResource("InputFiles/EmployeeList.txt"));
flatFileItemReader.setLineMapper(lineMapper());
return flatFileItemReader;
}
@Bean
public FlatFileItemWriter<Employee> flatFileWriter() {
BeanWrapperFieldExtractor<Employee> fieldExtractor = new BeanWrapperFieldExtractor<>();
fieldExtractor.setNames(new String[] {"id", "name", "location", "age"});
FileSystemResource fileSystemResource = new FileSystemResource("OutputFiles/employeeList.txt");
DelimitedLineAggregator<Employee> lineAggregator = new DelimitedLineAggregator<>();
lineAggregator.setDelimiter(",");
lineAggregator.setFieldExtractor(fieldExtractor);
return new FlatFileItemWriterBuilder<Employee>()
.name("flatFileItemWriter") //helps to uniquely recognize a job incase of need for retry
.resource(fileSystemResource)
.lineAggregator(lineAggregator)
.footerCallback(writer -> writer.write("Created @ " + new Date()))
.build();
}
@Bean
public FlatFileItemReader<EmployeeCSV> csvFileReader() {
return new FlatFileItemReaderBuilder<EmployeeCSV>()
.name("employeeCSVReader")
.resource(new FileSystemResource("EmployeeList.csv"))
.linesToSkip(1)
.delimited()
.names("ID", "Name", "Location", "Age")
.targetType(EmployeeCSV.class)
.build();
}
@Bean
public FlatFileItemWriter<Employee> csvFileWriter() {
BeanWrapperFieldExtractor<Employee> fieldExtractor = new BeanWrapperFieldExtractor<>();
fieldExtractor.setNames(new String[] {"id", "name", "location", "age"});
FileSystemResource fileSystemResource = new FileSystemResource("OutputFiles/EmployeeList.csv");
DelimitedLineAggregator<Employee> lineAggregator = new DelimitedLineAggregator<>();
lineAggregator.setDelimiter(",");
lineAggregator.setFieldExtractor(fieldExtractor);
return new FlatFileItemWriterBuilder<Employee>()
.name("csvItemWriter")
.resource(fileSystemResource)
.lineAggregator(lineAggregator)
.headerCallback(writer -> writer.write("ID,Name,Age,Location"))
.build();
}
@Bean
public JdbcCursorItemReader<EmployeeJdbc> jdbcReader() {
return new JdbcCursorItemReaderBuilder<EmployeeJdbc>()
.name("employeeJDBCReader")
.dataSource(dataSource)
.sql("SELECT id, name, age, location FROM public.tblemployee")
.rowMapper(new BeanPropertyRowMapper<>(EmployeeJdbc.class))
.build();
}
@Bean
public JdbcBatchItemWriter<Employee> jdbcWriter() {
return new JdbcBatchItemWriterBuilder<Employee>()
.itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
.sql("INSERT INTO tblemployee (id, name, location, age) VALUES (:id, :name, :location, :age)")
.dataSource(dataSource)
.build();
}
@Bean
public LineMapper<Employee> lineMapper() {
DefaultLineMapper<Employee> lineMapper = new DefaultLineMapper<>();
// Step 1: Tokenize the line
DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
tokenizer.setDelimiter(","); // Default is comma
tokenizer.setNames("ID", "Name", "Location", "Age"); // Must match CSV header
// Step 2: Map tokens to bean properties
BeanWrapperFieldSetMapper<Employee> fieldSetMapper = new BeanWrapperFieldSetMapper<>();
fieldSetMapper.setTargetType(Employee.class);
fieldSetMapper.setStrict(true); // Optional: disables fuzzy matching
// Step 3: Combine tokenizer and mapper
lineMapper.setLineTokenizer(tokenizer);
lineMapper.setFieldSetMapper(fieldSetMapper);
return lineMapper;
}
}
SimpleJobProcessor.java
import com.mugil.org.model.Employee;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.stereotype.Component;
@Component
public class SimpleJobProcessor implements ItemProcessor<Employee, Employee> {
@Override
public Employee process(Employee item) throws Exception {
System.out.println("Inside Job Processor");
String employeeDetails = "The Employee Name is " + item.getName() + ", his Age is " + item.getAge() + " and location is " + item.getLocation();
System.out.println(employeeDetails);
Employee objEmp = new Employee(item.getId(), item.getName(), item.getLocation(), item.getAge());
return objEmp;
}
}
