[New to SpringBatch] Using Spring Boot, I am trying to create a job which reads names from MongoDB, converts to lowercase, and outputs to CSV file. My reader and processor are working but the writer isn't.
My code is as follows.
Configuration file:
package bbye;
import java.util.HashMap;
import java.util.Map;
import javax.sql.DataSource;
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.data.MongoItemReader;
import org.springframework.batch.item.data.builder.MongoItemReaderBuilder;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor;
import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
import org.springframework.batch.item.file.transform.FieldExtractor;
import org.springframework.batch.item.file.transform.LineAggregator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import hello.Person;
@Configuration
@EnableBatchProcessing
public class BatchConfigProcessing {
@Autowired
public JobBuilderFactory jobBuilderFactory;
@Autowired
public StepBuilderFactory stepBuilderFactory;
@Autowired
private MongoTemplate mongoTemplate;
private String readQuery = "{}";
// tag::readerwriterprocessor[]
@Bean
public MongoItemReader<Person> readMongo(DataSource dataSource) {
return new MongoItemReaderBuilder<Person>()
.name("mongoDocReader")
.jsonQuery(readQuery)
.targetType(Person.class)
.sorts(sort())
.template(mongoTemplate)
.collection("people")
.build();
}
@Bean
public PersonDocProcessor processor() {
return new PersonDocProcessor();
}
@Bean
public FlatFileItemWriter<Person> writer() {
/*FlatFileItemWriterBuilder<Person> writePerson = new FlatFileItemWriterBuilder<Person>();
writePerson.name("personDocWriter");
writePerson.resource(new ClassPathResource("PersonExtracted.csv"));
writePerson.lineAggregator(new DelimitedLineAggregator<Person>());
writePerson.shouldDeleteIfExists(true);
writePerson.build();*/
FlatFileItemWriter<Person> fileWriter = new FlatFileItemWriter<>();
fileWriter.setName("csvWriter");
fileWriter.setResource(new ClassPathResource("PersonExtracted.csv"));
fileWriter.setLineAggregator(lineAggregator());
fileWriter.setForceSync(true);
fileWriter.close();
return fileWriter;
}
// end::readerwriterprocessor[]
// tag::jobstep[]
@Bean
public Job exportUserJob(FileUploadNotificationListener listener, Step step1) {
return jobBuilderFactory.get("exportUserJob")
.incrementer(new RunIdIncrementer())
.listener(listener)
.flow(step1)
.end()
.build();
}
@Bean
public Step step2(MongoItemReader<Person> reader) {
return stepBuilderFactory.get("step2")
.<Person, Person> chunk(10)
.reader(reader)
.processor(processor())
.writer(writer())
.build();
}
// end::jobstep[]
public FieldExtractor<Person> fieldExtractor()
{
BeanWrapperFieldExtractor<Person> extractor = new BeanWrapperFieldExtractor<>();
extractor.setNames( new String[] { "firstName",
"lastName"});
return extractor;
}
public LineAggregator<Person> lineAggregator() {
DelimitedLineAggregator<Person> la = new DelimitedLineAggregator<Person>();
la.setDelimiter(",");
la.setFieldExtractor(fieldExtractor());
return la;
}
public Map<String, Sort.Direction> sort(){
String firstName = "firstName";
Map<String, Sort.Direction> sortMap = new HashMap();
sortMap.put(firstName, Sort.DEFAULT_DIRECTION);
return sortMap;
}
}
Processor file
package bbye;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.stereotype.Component;
import hello.Person;
@Component
public class PersonDocProcessor implements ItemProcessor<Person, Person> {
private static final Logger log = LoggerFactory.getLogger(PersonDocProcessor.class);
@Override
public Person process(final Person person) throws Exception {
final String firstName = person.getFirstName().toLowerCase();
final String lastName = person.getLastName().toLowerCase();
final Person transformedPerson = new Person(firstName, lastName);
log.info("Converting (" + person + ") into (" + transformedPerson + ")");
return transformedPerson;
}
}
Listener
package bbye;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.stereotype.Component;
@Component
public class FileUploadNotificationListener implements JobExecutionListener {
@Override
public void beforeJob(JobExecution jobExecution) {
System.out.println("===== listening for job - mongoReader - fileWriter ====");
}
@Override
public void afterJob(JobExecution jobExecution) {
System.out.println("==== file write job completed =====");
}
}
Here Person is a simple POJO. The stack trace with and without manual file creation is as follows:
- If the file is not present under src/main/resources FlatFileItemWriter does not create a new file and throws
org.springframework.batch.item.ItemStreamException: Could not convert resource to file: [class path resource [PersonExtracted.csv]]
at org.springframework.batch.item.file.FlatFileItemWriter.getOutputState(FlatFileItemWriter.java:399) ~[spring-batch-infrastructure-4.0.1.RELEASE.jar:4.0.1.RELEASE]
at org.springframework.batch.item.file.FlatFileItemWriter.open(FlatFileItemWriter.java:337) ~[spring-batch-infrastructure-4.0.1.RELEASE.jar:4.0.1.RELEASE]
........
Caused by: java.io.FileNotFoundException: class path resource [PersonExtracted.csv] cannot be resolved to URL because it does not exist
- If I create the PersonExtracted.csv file manually, the program runs without errors but does not write anything to the file. In fact, a blank file is returned. The stack trace is as below.
:: Spring Boot :: (v2.0.2.RELEASE)
2018-06-19 11:35:17.663 INFO 25136 --- [ main] hello.Application : Starting Application on MyPC with PID 25136 (C:\eclipse-workspace\gs-batch-processing-master\complete\target\classes started by shristi in C:\eclipse-workspace\gs-batch-processing-master\complete)
2018-06-19 11:35:17.666 INFO 25136 --- [ main] hello.Application : No active profile set, falling back to default profiles: default
2018-06-19 11:35:17.689 INFO 25136 --- [ main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@15bb6bea: startup date [Tue Jun 19 11:35:17 EDT 2018]; root of context hierarchy
2018-06-19 11:35:18.135 INFO 25136 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2018-06-19 11:35:18.136 WARN 25136 --- [ main] com.zaxxer.hikari.util.DriverDataSource : Registered driver with driverClassName=org.hsqldb.jdbcDriver was not found, trying direct instantiation.
2018-06-19 11:35:18.282 INFO 25136 --- [ main] com.zaxxer.hikari.pool.PoolBase : HikariPool-1 - Driver does not support get/set network timeout for connections. (feature not supported)
2018-06-19 11:35:18.284 INFO 25136 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
2018-06-19 11:35:18.293 INFO 25136 --- [ main] o.s.jdbc.datasource.init.ScriptUtils : Executing SQL script from URL [file:/C:/eclipse-workspace/gs-batch-processing-master/complete/target/classes/schema-all.sql]
2018-06-19 11:35:18.297 INFO 25136 --- [ main] o.s.jdbc.datasource.init.ScriptUtils : Executed SQL script from URL [file:/C:/eclipse-workspace/gs-batch-processing-master/complete/target/classes/schema-all.sql] in 4 ms.
2018-06-19 11:35:18.518 INFO 25136 --- [ main] org.mongodb.driver.cluster : Cluster created with settings {hosts=[localhost:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2018-06-19 11:35:18.552 INFO 25136 --- [localhost:27017] org.mongodb.driver.connection : Opened connection [connectionId{localValue:1, serverValue:140}] to localhost:27017
2018-06-19 11:35:18.554 INFO 25136 --- [localhost:27017] org.mongodb.driver.cluster : Monitor thread successfully connected to server with description ServerDescription{address=localhost:27017, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 0]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=1438717}
2018-06-19 11:35:18.723 INFO 25136 --- [ main] o.s.b.c.r.s.JobRepositoryFactoryBean : No database type set, using meta data indicating: HSQL
2018-06-19 11:35:18.770 INFO 25136 --- [ main] o.s.b.c.l.support.SimpleJobLauncher : No TaskExecutor has been set, defaulting to synchronous executor.
2018-06-19 11:35:18.778 INFO 25136 --- [ main] o.s.jdbc.datasource.init.ScriptUtils : Executing SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql]
2018-06-19 11:35:18.781 INFO 25136 --- [ main] o.s.jdbc.datasource.init.ScriptUtils : Executed SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql] in 3 ms.
2018-06-19 11:35:18.870 INFO 25136 --- [ main] o.s.j.e.a.AnnotationMBeanExporter : Registering beans for JMX exposure on startup
2018-06-19 11:35:18.871 INFO 25136 --- [ main] o.s.j.e.a.AnnotationMBeanExporter : Bean with name 'dataSource' has been autodetected for JMX exposure
2018-06-19 11:35:18.873 INFO 25136 --- [ main] o.s.j.e.a.AnnotationMBeanExporter : Located MBean 'dataSource': registering with JMX server as MBean [com.zaxxer.hikari:name=dataSource,type=HikariDataSource]
2018-06-19 11:35:18.880 INFO 25136 --- [ main] hello.Application : Started Application in 1.357 seconds (JVM running for 2.284)
2018-06-19 11:35:18.881 INFO 25136 --- [ main] o.s.b.a.b.JobLauncherCommandLineRunner : Running default command line with: []
2018-06-19 11:35:18.908 INFO 25136 --- [ main] o.s.b.c.l.support.SimpleJobLauncher : Job: [FlowJob: [name=exportUserJob]] launched with the following parameters: [{run.id=1}]
===== listening for job - mongoReader - fileWriter ====
2018-06-19 11:35:18.917 INFO 25136 --- [ main] o.s.batch.core.job.SimpleStepHandler : Executing step: [step2]
2018-06-19 11:35:18.995 INFO 25136 --- [ main] org.mongodb.driver.connection : Opened connection [connectionId{localValue:2, serverValue:141}] to localhost:27017
2018-06-19 11:35:19.022 INFO 25136 --- [ main] bbye.PersonDocProcessor : Converting (firstName: ALICE, lastName: WONDERLAND) into (firstName: alice, lastName: wonderland)
2018-06-19 11:35:19.022 INFO 25136 --- [ main] bbye.PersonDocProcessor : Converting (firstName: FIRSTNAME, lastName: LASTNAME) into (firstName: firstname, lastName: lastname)
2018-06-19 11:35:19.022 INFO 25136 --- [ main] bbye.PersonDocProcessor : Converting (firstName: JANE, lastName: DOE) into (firstName: jane, lastName: doe)
2018-06-19 11:35:19.022 INFO 25136 --- [ main] bbye.PersonDocProcessor : Converting (firstName: JOHN, lastName: DOE) into (firstName: john, lastName: doe)
2018-06-19 11:35:19.022 INFO 25136 --- [ main] bbye.PersonDocProcessor : Converting (firstName: MARK, lastName: WINN) into (firstName: mark, lastName: winn)
==== file write job completed =====
2018-06-19 11:35:19.031 INFO 25136 --- [ main] o.s.b.c.l.support.SimpleJobLauncher : Job: [FlowJob: [name=exportUserJob]] completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]
2018-06-19 11:35:19.032 INFO 25136 --- [ Thread-2] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@15bb6bea: startup date [Tue Jun 19 11:35:17 EDT 2018]; root of context hierarchy
2018-06-19 11:35:19.033 INFO 25136 --- [ Thread-2] o.s.j.e.a.AnnotationMBeanExporter : Unregistering JMX-exposed beans on shutdown
2018-06-19 11:35:19.034 INFO 25136 --- [ Thread-2] o.s.j.e.a.AnnotationMBeanExporter : Unregistering JMX-exposed beans
2018-06-19 11:35:19.035 INFO 25136 --- [ Thread-2] org.mongodb.driver.connection : Closed connection [connectionId{localValue:2, serverValue:141}] to localhost:27017 because the pool has been closed.
2018-06-19 11:35:19.036 INFO 25136 --- [ Thread-2] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Shutdown initiated...
2018-06-19 11:35:19.037 INFO 25136 --- [ Thread-2] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Shutdown completed.