Create basic batch ingest sample
* Read from a file, apply processing, write to database * Hook in Spring Cloud Task * Ensure job can be registered and executed through data flow * Create README documenting build steps and how to register / launch the job as a task via Data Flow shell
This commit is contained in:
Chris Schaefer
committed by
David Turanski
David Turanski
parent
02fc1d308a
commit
b68aaf2487
@@ -0,0 +1,18 @@
|
||||
package org.springframework.ingest;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.task.configuration.EnableTask;
|
||||
|
||||
/**
|
||||
* Main entry point for the ingest sample application.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
@EnableTask
|
||||
@SpringBootApplication
|
||||
public class Application {
|
||||
public static void main(String[] args) throws Exception {
|
||||
SpringApplication.run(Application.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package org.springframework.ingest.config;
|
||||
|
||||
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.configuration.annotation.StepScope;
|
||||
import org.springframework.batch.core.launch.support.RunIdIncrementer;
|
||||
import org.springframework.batch.item.ItemProcessor;
|
||||
import org.springframework.batch.item.ItemStreamReader;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
|
||||
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.io.ResourceLoader;
|
||||
|
||||
import org.springframework.ingest.domain.Person;
|
||||
import org.springframework.ingest.mapper.fieldset.PersonFieldSetMapper;
|
||||
import org.springframework.ingest.processor.PersonItemProcessor;
|
||||
|
||||
/**
|
||||
* Class used to configure the batch job related beans.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
public class BatchConfiguration {
|
||||
private final DataSource dataSource;
|
||||
private final ResourceLoader resourceLoader;
|
||||
private final JobBuilderFactory jobBuilderFactory;
|
||||
private final StepBuilderFactory stepBuilderFactory;
|
||||
|
||||
@Autowired
|
||||
public BatchConfiguration(final DataSource dataSource, final JobBuilderFactory jobBuilderFactory,
|
||||
final StepBuilderFactory stepBuilderFactory,
|
||||
final ResourceLoader resourceLoader) {
|
||||
this.dataSource = dataSource;
|
||||
this.resourceLoader = resourceLoader;
|
||||
this.jobBuilderFactory = jobBuilderFactory;
|
||||
this.stepBuilderFactory = stepBuilderFactory;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@StepScope
|
||||
public ItemStreamReader<Person> reader(@Value("#{jobParameters['filePath']}") String filePath) throws Exception {
|
||||
return new FlatFileItemReaderBuilder<Person>()
|
||||
.name("reader")
|
||||
.resource(resourceLoader.getResource(filePath))
|
||||
.delimited()
|
||||
.names(new String[] {"firstName", "lastName"})
|
||||
.fieldSetMapper(new PersonFieldSetMapper())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ItemProcessor<Person, Person> processor() {
|
||||
return new PersonItemProcessor();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ItemWriter<Person> writer() {
|
||||
return new JdbcBatchItemWriterBuilder<Person>()
|
||||
.beanMapped()
|
||||
.dataSource(this.dataSource)
|
||||
.sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Job ingestJob() throws Exception {
|
||||
return jobBuilderFactory.get("ingestJob")
|
||||
.incrementer(new RunIdIncrementer())
|
||||
.flow(step1())
|
||||
.end()
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Step step1() throws Exception {
|
||||
return stepBuilderFactory.get("ingest")
|
||||
.<Person, Person>chunk(10)
|
||||
.reader(reader(null))
|
||||
.processor(processor())
|
||||
.writer(writer())
|
||||
.build();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package org.springframework.ingest.domain;
|
||||
|
||||
/**
|
||||
* Domain object representing data about a Person.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
public class Person {
|
||||
private final String firstName;
|
||||
private final String lastName;
|
||||
|
||||
public Person(final String firstName, final String lastName) {
|
||||
this.firstName = firstName;
|
||||
this.lastName = lastName;
|
||||
}
|
||||
|
||||
public String getFirstName() {
|
||||
return firstName;
|
||||
}
|
||||
|
||||
public String getLastName() {
|
||||
return lastName;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "First name: " + firstName + " , last name: " + lastName;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package org.springframework.ingest.mapper.fieldset;
|
||||
|
||||
import org.springframework.batch.item.file.mapping.FieldSetMapper;
|
||||
import org.springframework.batch.item.file.transform.FieldSet;
|
||||
|
||||
import org.springframework.ingest.domain.Person;
|
||||
|
||||
/**
|
||||
* Maps the provided FieldSet into a Person object.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
public class PersonFieldSetMapper implements FieldSetMapper<Person> {
|
||||
@Override
|
||||
public Person mapFieldSet(FieldSet fieldSet) {
|
||||
String firstName = fieldSet.readString(0);
|
||||
String lastName = fieldSet.readString(1);
|
||||
|
||||
return new Person(firstName, lastName);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package org.springframework.ingest.processor;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.springframework.batch.item.ItemProcessor;
|
||||
import org.springframework.ingest.domain.Person;
|
||||
|
||||
/**
|
||||
* Processes the providing record, transforming the data into
|
||||
* uppercase characters.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
public class PersonItemProcessor implements ItemProcessor<Person, Person> {
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(PersonItemProcessor.class);
|
||||
|
||||
@Override
|
||||
public Person process(Person person) throws Exception {
|
||||
String firstName = person.getFirstName().toUpperCase();
|
||||
String lastName = person.getLastName().toUpperCase();
|
||||
|
||||
Person processedPerson = new Person(firstName, lastName);
|
||||
|
||||
LOGGER.info("Processed: " + person + " into: " + processedPerson);
|
||||
|
||||
return processedPerson;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
spring.application.name=fileIngest
|
||||
5
batch/file-ingest/src/main/resources/data.csv
Normal file
5
batch/file-ingest/src/main/resources/data.csv
Normal file
@@ -0,0 +1,5 @@
|
||||
Jill,Doe
|
||||
Joe,Doe
|
||||
Justin,Doe
|
||||
Jane,Doe
|
||||
John,Doe
|
||||
|
7
batch/file-ingest/src/main/resources/schema-all.sql
Normal file
7
batch/file-ingest/src/main/resources/schema-all.sql
Normal file
@@ -0,0 +1,7 @@
|
||||
DROP TABLE people IF EXISTS;
|
||||
|
||||
CREATE TABLE people (
|
||||
person_id BIGINT IDENTITY NOT NULL PRIMARY KEY,
|
||||
first_name VARCHAR(20),
|
||||
last_name VARCHAR(20)
|
||||
);
|
||||
@@ -0,0 +1,120 @@
|
||||
package org.springframework.ingest.config;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.core.JobParameters;
|
||||
import org.springframework.batch.core.JobParametersBuilder;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.io.ResourceLoader;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseFactory;
|
||||
import org.springframework.jdbc.datasource.init.DatabasePopulatorUtils;
|
||||
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
|
||||
/**
|
||||
* BatchConfiguration test cases
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
public class BatchConfigurationTests {
|
||||
private static AnnotationConfigApplicationContext context;
|
||||
|
||||
@Before
|
||||
public void createContext() {
|
||||
context = new AnnotationConfigApplicationContext(new Class[] {
|
||||
BatchConfiguration.class, BatchConfigurationTests.DataSourceConfiguration.class });
|
||||
}
|
||||
|
||||
@After
|
||||
public void closeContext() {
|
||||
context.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBatchConfigurationSuccess() throws Exception {
|
||||
JobExecution jobExecution = testJob("classpath:data.csv");
|
||||
|
||||
assertEquals("Incorrect batch status", BatchStatus.COMPLETED, jobExecution.getStatus());
|
||||
assertEquals("Invalid number of step executions", 1, jobExecution.getStepExecutions().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBatchConfigurationFail() throws Exception {
|
||||
JobExecution jobExecution = testJob("classpath:missing-data-file.csv");
|
||||
|
||||
assertEquals("Incorrect batch status", BatchStatus.FAILED, jobExecution.getStatus());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBatchDataProcessing() throws Exception {
|
||||
JobExecution jobExecution = testJob("classpath:data.csv");
|
||||
|
||||
assertEquals("Incorrect batch status", BatchStatus.COMPLETED, jobExecution.getStatus());
|
||||
assertEquals("Invalid number of step executions", 1, jobExecution.getStepExecutions().size());
|
||||
|
||||
JdbcTemplate jdbcTemplate = new JdbcTemplate(context.getBean(DataSource.class));
|
||||
List<Map<String, Object>> peopleList = jdbcTemplate.queryForList("select first_name, last_name from people");
|
||||
|
||||
assertEquals("Incorrect number of results", 5, peopleList.size());
|
||||
|
||||
for(Map<String, Object> person : peopleList) {
|
||||
assertNotNull("Received null person", person);
|
||||
|
||||
String firstName = (String) person.get("first_name");
|
||||
assertEquals("Invalid first name: " + firstName, firstName.toUpperCase(), firstName);
|
||||
|
||||
String lastName = (String) person.get("last_name");
|
||||
assertEquals("Invalid last name: " + lastName, lastName.toUpperCase(), lastName);
|
||||
}
|
||||
}
|
||||
|
||||
private JobExecution testJob(String filePath) throws Exception {
|
||||
Job job = context.getBean(Job.class);
|
||||
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
|
||||
JobParameters jobParameters = new JobParametersBuilder()
|
||||
.addString("filePath", filePath)
|
||||
.toJobParameters();
|
||||
|
||||
return jobLauncher.run(job, jobParameters);
|
||||
}
|
||||
|
||||
@Configuration
|
||||
public static class DataSourceConfiguration {
|
||||
@Autowired
|
||||
private ResourceLoader resourceLoader;
|
||||
|
||||
@PostConstruct
|
||||
protected void initialize() {
|
||||
ResourceDatabasePopulator populator = new ResourceDatabasePopulator();
|
||||
populator.addScript(resourceLoader.getResource(ClassUtils.addResourcePathToPackagePath(Step.class, "schema-hsqldb.sql")));
|
||||
populator.addScript(resourceLoader.getResource("classpath:schema-all.sql"));
|
||||
populator.setContinueOnError(true);
|
||||
DatabasePopulatorUtils.execute(populator, dataSource());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DataSource dataSource() {
|
||||
return new EmbeddedDatabaseFactory().getDatabase();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package org.springframework.ingest.mapper.fieldset;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.batch.item.file.mapping.FieldSetMapper;
|
||||
import org.springframework.batch.item.file.transform.DefaultFieldSet;
|
||||
import org.springframework.batch.item.file.transform.FieldSet;
|
||||
|
||||
import org.springframework.ingest.domain.Person;
|
||||
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/**
|
||||
* Test cases for PersonFieldSetMapper.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
public class PersonFieldSetMapperTests {
|
||||
private static final String[] TOKENS = new String[] { "jane", "doe" };
|
||||
private static final String[] NAMES = new String[] { "firstName", "lastName" };
|
||||
|
||||
@Test
|
||||
public void testPersonFieldMapping() throws Exception {
|
||||
FieldSet fieldSet = new DefaultFieldSet(TOKENS, NAMES);
|
||||
|
||||
FieldSetMapper<Person> fieldSetMapper = new PersonFieldSetMapper();
|
||||
Person person = fieldSetMapper.mapFieldSet(fieldSet);
|
||||
|
||||
assertNotNull("Received null Person", person);
|
||||
assertNotNull("Received null first name", person.getFirstName());
|
||||
assertNotNull("Received null last name", person.getLastName());
|
||||
assertEquals("Received wrong first name", TOKENS[0], person.getFirstName());
|
||||
assertEquals("Received wrong last name", TOKENS[1], person.getLastName());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package org.springframework.ingest.processor;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.batch.item.ItemProcessor;
|
||||
import org.springframework.ingest.domain.Person;
|
||||
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/**
|
||||
* Test cases for PersonItemProcessor.
|
||||
*
|
||||
* @author Chris Schaefer
|
||||
*/
|
||||
public class PersonItemProcessorTests {
|
||||
private static final String FIRST_NAME = "jane";
|
||||
private static final String LAST_NAME = "doe";
|
||||
|
||||
@Test
|
||||
public void testPersonProcessing() throws Exception {
|
||||
Person person = new Person(FIRST_NAME, LAST_NAME);
|
||||
|
||||
ItemProcessor<Person, Person> personItemProcessor = new PersonItemProcessor();
|
||||
Person transformedPerson = personItemProcessor.process(person);
|
||||
|
||||
assertNotNull("Received null Person", transformedPerson);
|
||||
assertNotNull("Received null first name", transformedPerson.getFirstName());
|
||||
assertNotNull("Received null last name", transformedPerson.getLastName());
|
||||
assertEquals("Invalid first name processing, should be uppercase",
|
||||
person.getFirstName().toUpperCase(), transformedPerson.getFirstName());
|
||||
assertEquals("Invalid last name processing, should be uppercase",
|
||||
person.getLastName().toUpperCase(), transformedPerson.getLastName());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user