diff --git a/batch/file-ingest-sftp/data/people.csv b/batch/file-ingest-sftp/data/people.csv new file mode 100644 index 0000000..5f3258e --- /dev/null +++ b/batch/file-ingest-sftp/data/people.csv @@ -0,0 +1,3 @@ +Jane,Doe +John,Doe +Joe,Doe \ No newline at end of file diff --git a/batch/file-ingest-sftp/pom.xml b/batch/file-ingest-sftp/pom.xml new file mode 100644 index 0000000..af47cb9 --- /dev/null +++ b/batch/file-ingest-sftp/pom.xml @@ -0,0 +1,117 @@ + + + 4.0.0 + org.springframework + ingest-sftp + 1.0.0 + + org.springframework.boot + spring-boot-starter-parent + 2.0.1.RELEASE + + + 1.8 + 3.7.0 + 1.2.2.RELEASE + 5.0.4.RELEASE + 2.4 + 1.6.0 + + + + org.springframework.boot + spring-boot-starter-batch + + + com.h2database + h2 + + + org.springframework.cloud + spring-cloud-task-core + ${spring.cloud.task.version} + + + org.springframework.cloud + spring-cloud-task-batch + ${spring.cloud.task.version} + + + commons-io + commons-io + ${commons.io.version} + + + org.springframework.integration + spring-integration-file + ${spring.integration.version} + + + org.springframework.integration + spring-integration-sftp + ${spring.integration.version} + + + org.apache.sshd + sshd-core + ${sshd.core.version} + test + + + junit + junit + test + + + org.hsqldb + hsqldb + test + + + org.springframework.boot + spring-boot-test + + + org.springframework + spring-test + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-compiler-plugin + ${maven.compiler.plugin.version} + + ${java.version} + ${java.version} + ${java.version} + ${java.version} + -Xlint:all + + + + + + + repository.spring.milestone + Spring Milestone Repository + http://repo.spring.io/milestone + + + + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + + false + + + + diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/Application.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/Application.java new file mode 100644 index 0000000..1cbaf9d --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/Application.java @@ -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); + } +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfiguration.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfiguration.java new file mode 100644 index 0000000..dffc988 --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfiguration.java @@ -0,0 +1,137 @@ +package org.springframework.ingest.config; + +import java.io.File; + +import javax.sql.DataSource; + +import org.apache.commons.io.FileUtils; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.StepExecutionListener; +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.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; + +import org.springframework.ingest.domain.Person; +import org.springframework.ingest.mapper.fieldset.PersonFieldSetMapper; +import org.springframework.ingest.processor.PersonItemProcessor; +import org.springframework.ingest.resource.RemoteResource; +import org.springframework.ingest.resource.sftp.SftpRemoteResource; + +/** + * Class used to configure the batch job related beans. + * + * @author Chris Schaefer + */ +@Configuration +@EnableBatchProcessing +@EnableConfigurationProperties(BatchConfigurationProperties.class) +public class BatchConfiguration { + private final DataSource dataSource; + private final JobBuilderFactory jobBuilderFactory; + private final StepBuilderFactory stepBuilderFactory; + private final BatchConfigurationProperties batchConfigurationProperties; + + @Autowired + public BatchConfiguration(final DataSource dataSource, final JobBuilderFactory jobBuilderFactory, + final StepBuilderFactory stepBuilderFactory, + final BatchConfigurationProperties batchConfigurationProperties) { + this.dataSource = dataSource; + this.jobBuilderFactory = jobBuilderFactory; + this.stepBuilderFactory = stepBuilderFactory; + this.batchConfigurationProperties = batchConfigurationProperties; + } + + @Bean + @StepScope + public StepExecutionListener ingestStepExecutionListener(@Value("#{jobParameters['remoteFilePath']}") String remoteFilePath, + @Value("#{jobParameters['localFilePath']}") String localFilePath) { + return new StepExecutionListener() { + @Override + public void beforeStep(StepExecution stepExecution) { + try { + Resource fetchedResource = remoteResource().getResource(remoteFilePath); + FileUtils.copyInputStreamToFile(fetchedResource.getInputStream(), new File(localFilePath)); + } + catch (Exception e) { + throw new RuntimeException("Could not write remote file to local disk", e); + } + } + + @Override + public ExitStatus afterStep(StepExecution stepExecution) { + return null; + } + }; + } + + @Bean + @StepScope + public ItemStreamReader reader(@Value("#{jobParameters['localFilePath']}") String localFilePath) throws Exception { + return new FlatFileItemReaderBuilder() + .name("reader") + .resource(new FileSystemResource(localFilePath)) + .delimited() + .names(new String[] {"firstName", "lastName"}) + .fieldSetMapper(new PersonFieldSetMapper()) + .build(); + } + + @Bean + public ItemProcessor processor() { + return new PersonItemProcessor(); + } + + @Bean + public ItemWriter writer() { + return new JdbcBatchItemWriterBuilder() + .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") + .chunk(10) + .reader(reader(null)) + .processor(processor()) + .writer(writer()) + .listener(ingestStepExecutionListener(null, null)) + .build(); + } + + @Bean + public RemoteResource remoteResource() { + return new SftpRemoteResource(batchConfigurationProperties.getSftpHost(), batchConfigurationProperties.getSftpPort(), + batchConfigurationProperties.getSftpUsername(), batchConfigurationProperties.getSftpPassword()); + } +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfigurationProperties.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfigurationProperties.java new file mode 100644 index 0000000..b5db31f --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/config/BatchConfigurationProperties.java @@ -0,0 +1,48 @@ +package org.springframework.ingest.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Configuration Properties for BatchConfiguration + * + * @author Chris Schaefer + */ +@ConfigurationProperties +public class BatchConfigurationProperties { + private String sftpHost; + private Integer sftpPort; + private String sftpUsername; + private String sftpPassword; + + public String getSftpHost() { + return sftpHost; + } + + public void setSftpHost(String sftpHost) { + this.sftpHost = sftpHost; + } + + public Integer getSftpPort() { + return sftpPort; + } + + public void setSftpPort(Integer sftpPort) { + this.sftpPort = sftpPort; + } + + public String getSftpUsername() { + return sftpUsername; + } + + public void setSftpUsername(String sftpUsername) { + this.sftpUsername = sftpUsername; + } + + public String getSftpPassword() { + return sftpPassword; + } + + public void setSftpPassword(String sftpPassword) { + this.sftpPassword = sftpPassword; + } +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/domain/Person.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/domain/Person.java new file mode 100644 index 0000000..edf03d9 --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/domain/Person.java @@ -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; + } +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapper.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapper.java new file mode 100644 index 0000000..4b3f6cf --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapper.java @@ -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 { + @Override + public Person mapFieldSet(FieldSet fieldSet) { + String firstName = fieldSet.readString(0); + String lastName = fieldSet.readString(1); + + return new Person(firstName, lastName); + } +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/processor/PersonItemProcessor.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/processor/PersonItemProcessor.java new file mode 100644 index 0000000..6d2802c --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/processor/PersonItemProcessor.java @@ -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 { + 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; + } +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/RemoteResource.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/RemoteResource.java new file mode 100644 index 0000000..01da55e --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/RemoteResource.java @@ -0,0 +1,12 @@ +package org.springframework.ingest.resource; + +import org.springframework.core.io.Resource; + +/** + * Interface definition for remote Resource implementations. + * + * @author Chris Schaefer + */ +public interface RemoteResource { + Resource getResource(String resourceLocation); +} diff --git a/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/sftp/SftpRemoteResource.java b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/sftp/SftpRemoteResource.java new file mode 100644 index 0000000..2fea44b --- /dev/null +++ b/batch/file-ingest-sftp/src/main/java/org/springframework/ingest/resource/sftp/SftpRemoteResource.java @@ -0,0 +1,81 @@ +package org.springframework.ingest.resource.sftp; + +import java.io.InputStream; + +import org.apache.commons.io.IOUtils; + +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +import org.springframework.ingest.resource.RemoteResource; +import org.springframework.integration.file.remote.InputStreamCallback; +import org.springframework.integration.file.remote.RemoteFileOperations; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; +import org.springframework.util.Assert; + +/** + * RemoteResource implementation utilizing a Spring Integration SftpRemoteFileTemplate + * to connect and return a file as a Spring Resource. + * + * @author Chris Schaefer + */ +public class SftpRemoteResource implements RemoteResource { + private static final Integer DEFAULT_SFTP_PORT = 22; + private static final String LOCALHOST = "127.0.0.1"; + + private final String host; + private final Integer port; + private final String username; + private final String password; + + public SftpRemoteResource(String host, Integer port, String username, String password) { + Assert.hasText(username, "Username must be defined"); + Assert.hasText(password, "Password must be defined"); + + this.host = host; + this.port = port; + this.username = username; + this.password = password; + } + + @Override + public Resource getResource(String resourceLocation) { + DefaultSftpSessionFactory sessionFactory = getSessionFactory(); + RemoteFileOperations remoteFileOperations = new SftpRemoteFileTemplate(sessionFactory); + + FileFetcher filefetcher = new FileFetcher(); + remoteFileOperations.get(resourceLocation, filefetcher); + + return new ByteArrayResource(filefetcher.getBytes()); + } + + private DefaultSftpSessionFactory getSessionFactory() { + DefaultSftpSessionFactory sessionFactory = new DefaultSftpSessionFactory(); + sessionFactory.setHost(host != null ? host : LOCALHOST); + sessionFactory.setPort(port != null ? port : DEFAULT_SFTP_PORT); + sessionFactory.setUser(username); + sessionFactory.setPassword(password); + sessionFactory.setAllowUnknownKeys(true); + + return sessionFactory; + } + + + private static class FileFetcher implements InputStreamCallback { + private byte[] bytes; + + @Override + public void doWithInputStream(InputStream inputStream) { + try { + bytes = IOUtils.toByteArray(inputStream); + } + catch (Exception e) { + throw new RuntimeException("Failed to convert InputStream to byte array", e); + } + } + + public byte[] getBytes() { + return bytes; + } + } +} diff --git a/batch/file-ingest-sftp/src/main/resources/application.properties b/batch/file-ingest-sftp/src/main/resources/application.properties new file mode 100644 index 0000000..6986449 --- /dev/null +++ b/batch/file-ingest-sftp/src/main/resources/application.properties @@ -0,0 +1 @@ +spring.application.name=fileIngestSftp diff --git a/batch/file-ingest-sftp/src/main/resources/schema-all.sql b/batch/file-ingest-sftp/src/main/resources/schema-all.sql new file mode 100644 index 0000000..4e798d0 --- /dev/null +++ b/batch/file-ingest-sftp/src/main/resources/schema-all.sql @@ -0,0 +1,5 @@ +CREATE TABLE IF NOT EXISTS people ( + person_id BIGINT IDENTITY NOT NULL PRIMARY KEY, + first_name VARCHAR(20), + last_name VARCHAR(20) +); diff --git a/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/config/BatchConfigurationTests.java b/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/config/BatchConfigurationTests.java new file mode 100644 index 0000000..8a02f43 --- /dev/null +++ b/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/config/BatchConfigurationTests.java @@ -0,0 +1,202 @@ +package org.springframework.ingest.config; + +import org.junit.runner.RunWith; +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.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.env.ConfigurableEnvironment; +import org.springframework.core.env.MapPropertySource; +import org.springframework.core.env.StandardEnvironment; +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.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.ClassUtils; + +import org.apache.commons.io.FileUtils; + +import org.apache.sshd.common.NamedFactory; +import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory; +import org.apache.sshd.server.Command; +import org.apache.sshd.server.SshServer; +import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; +import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory; + +import java.io.File; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.HashMap; + +import javax.annotation.PostConstruct; +import javax.sql.DataSource; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.rules.TemporaryFolder; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * BatchConfiguration test cases + * + * @author Chris Schaefer + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE) +@DirtiesContext +@EnableConfigurationProperties(BatchConfigurationProperties.class) +public class BatchConfigurationTests { + private static int port; + private static SshServer server; + private static AnnotationConfigApplicationContext context; + + private static final String SFTP_USER = "user"; + private static final String SFTP_PASS = "pass"; + private static final String SFTP_HOST = "127.0.0.1"; + private static final String REMOTE_FILE = "people.csv"; + private static final String HOST_KEY_FILE = "hostkey.ser"; + + @ClassRule + public static final TemporaryFolder remoteTemporaryFolder = new TemporaryFolder(); + + @BeforeClass + public static void createServer() throws Exception { + File createdFile = remoteTemporaryFolder.newFile(REMOTE_FILE); + FileUtils.writeStringToFile(createdFile, "Jill,Doe\nJoe,Doe\nJustin,Doe\nJane,Doe\nJohn,Doe"); + + server = SshServer.setUpDefaultServer(); + server.setPasswordAuthenticator((username, password, session) -> true); + server.setPort(0); + server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File(HOST_KEY_FILE))); + server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystemFactory())); + server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder.getRoot().toPath())); + server.start(); + + port = server.getPort(); + } + + @AfterClass + public static void stopServer() throws Exception { + server.stop(); + + File hostkey = new File(HOST_KEY_FILE); + + if (hostkey.exists()) { + hostkey.delete(); + } + } + + @Before + public void createContext() { + Map properties = new HashMap(); + properties.put("sftp_host", SFTP_HOST); + properties.put("sftp_port", port); + properties.put("sftp_username", SFTP_USER); + properties.put("sftp_password", SFTP_PASS); + + ConfigurableEnvironment environment = new StandardEnvironment(); + environment.getPropertySources().addFirst(new MapPropertySource("sftpProperties", properties)); + + context = new AnnotationConfigApplicationContext(); + context.register(BatchConfiguration.class, DataSourceConfiguration.class); + context.setEnvironment(environment); + context.refresh(); + } + + @After + public void closeContext() { + context.close(); + } + + @Test + public void testBatchConfigurationSuccess() throws Exception { + JobExecution jobExecution = testJob(REMOTE_FILE); + + 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("missing-people-file.csv"); + + assertEquals("Incorrect batch status", BatchStatus.FAILED, jobExecution.getStatus()); + } + + @Test + public void testBatchDataProcessing() throws Exception { + JobExecution jobExecution = testJob(REMOTE_FILE); + + 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> peopleList = jdbcTemplate.queryForList("select first_name, last_name from people"); + + assertEquals("Incorrect number of results", 5, peopleList.size()); + + for(Map 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); + + File localFile = File.createTempFile("local", ".csv"); + localFile.deleteOnExit(); + + JobParameters jobParameters = new JobParametersBuilder() + .addString("localFilePath", localFile.getAbsolutePath()) + .addString("remoteFilePath", 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(); + } + } +} diff --git a/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapperTests.java b/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapperTests.java new file mode 100644 index 0000000..4256de3 --- /dev/null +++ b/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/mapper/fieldset/PersonFieldSetMapperTests.java @@ -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 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()); + } +} diff --git a/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/processor/PersonItemProcessorTests.java b/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/processor/PersonItemProcessorTests.java new file mode 100644 index 0000000..1477f58 --- /dev/null +++ b/batch/file-ingest-sftp/src/test/java/org/springframework/ingest/processor/PersonItemProcessorTests.java @@ -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 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()); + } +} diff --git a/src/main/asciidoc/batch/file-ingest-sftp/main.adoc b/src/main/asciidoc/batch/file-ingest-sftp/main.adoc new file mode 100644 index 0000000..8c341f6 --- /dev/null +++ b/src/main/asciidoc/batch/file-ingest-sftp/main.adoc @@ -0,0 +1,236 @@ + +=== Batch File Ingest - SFTP + +In the <> demonstration we built a link:https://projects.spring.io/spring-batch[Spring Batch] application that would deploy into link:https://cloud.spring.io/spring-cloud-dataflow[Spring Cloud Data Flow] as a task and process a file embedded in the batch job JAR. This time we will build upon that sample but rather than deploying as a task in Spring Cloud Data Flow, we will create a link:https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#spring-cloud-dataflow-streams[Stream]. This stream will poll an SFTP server and for each new file that it finds it launches the batch job to download the file and process it. + +==== Prerequisites + +* Running instance of link:http://kafka.apache.org/downloads.html[Kafka] + +* Running instance of link:https://redis.io/download[Redis] + +* A Running Data Flow Server +include::{docs_dir}/local-server.adoc[] + +* A Running Data Flow Shell +include::{docs_dir}/shell.adoc[] + +* Either a remote or local host accepting SFTP connections. + +* A database tool such as link:https://dbeaver.jkiss.org/download/[DBeaver] to inspect the database contents + +NOTE: To simplify the dependencies and configuration in this example, we will use our local machine acting as an SFTP server. + +==== Batch File Ingest SFTP Demo Overview + +The source for the demo project is located in the `batch/file-ingest-sftp` directory at the top-level of this repository. The code in this directory is built upon the same code found in <>. + +The key modifications from the <> sample are: + +* `BatchConfiguration` - The main change to this class is the addition of a link:https://docs.spring.io/spring-batch/trunk/apidocs/org/springframework/batch/core/StepExecutionListener.html[`StepExecutionLister`]. This listener gets set into the configuration of `step1` so on execution of the step, the listener will fetch the provided file. Since we are now fetching a file from a remote resource and downloading it for processing, we obtain the remote file location from a link:https://docs.spring.io/spring-batch/trunk/apidocs/org/springframework/batch/core/JobParameter.html[`JobParameter`] named `remoteFilePath` and the path to where the downloaded file will be stored as under the `JobParameter` named `localFilePath`. + +* `SftpRemoteResource` - To obtain the file from SFTP, this class was created utilizing the link:https://docs.spring.io/spring-integration/api/org/springframework/integration/sftp/session/SftpRemoteFileTemplate.html[`SftpRemoteFileTemplate`] class from link:https://projects.spring.io/spring-integration/[Spring Integration]. We create a new bean from this class in `BatchConfiguration` and the `StepExecutionListener` uses it to fetch files. + +Additionally we use Redis to persist the paths of files we have seen on the SFTP server. We persist this data rather than storing it in memory so the the seen files won't be sent for processing in the event of a failure. + +==== Building and Running the Demo + +. Build the demo JAR ++ +From the root of this project: ++ +``` +$ cd batch/file-ingest-sftp +$ mvn clean package +``` ++ + +. Create the data directories ++ +Now we create directories where the batch job expects to find files that would be on the remote SFTP server as well as where they should be transferred locally. These paths must exist prior to running the batch job. ++ +NOTE: If you are using a non-local SFTP server, the `/tmp/remote-files` directory would be created on the SFTP server and `/tmp/local-files` would be created on your local machine. ++ +``` +$ mkdir -p /tmp/remote-files /tmp/local-files +``` ++ + +. Register the the SFTP source and the Task Launcher Local sink ++ +With our Spring Cloud Data Flow server running, we register the `SFTP` source and `task-launcher-local` sink. The `SFTP` source application will do the work of polling for new files and when received, it sends a message to the `task-launcher-local` to launch the batch job for that file. ++ +In the Spring Cloud Data Flow shell: ++ +[source,console,options=nowrap] +---- +dataflow:>app register --name sftp --type source --uri maven://org.springframework.cloud.stream.app:sftp-source-kafka:2.0.0.BUILD-SNAPSHOT +Successfully registered application 'source:sftp' +dataflow:>app register --name task-launcher-local --type sink --uri maven://org.springframework.cloud.stream.app:task-launcher-local-sink-kafka:2.0.0.M1 +Successfully registered application 'sink:task-launcher-local' +---- ++ + +. Create and deploy the stream ++ +Now lets create and deploy the stream which will start polling the SFTP server and when new files arrive launch the batch job. ++ +NOTE: you must replace `--username=user`, `--password=pass` and `--batch-resource-uri=file:////path/to/sftp-ingest.jar` below to their respective values. The `--username=` and `--password=` parameters are the credentials for your local (or remote) user and `--batch-resource-uri=` is the fully qualified path to the sample ingest JAR we built above. If rather than using the local system as an SFTP server, to specify the host use the `--host=` parameter and optionally `--port=`. If not defined, `--host` defaults to `127.0.0.1` and port defaults to `22`. ++ +[source,console,options=nowrap] +---- +dataflow:>stream create --name inboundSftp --definition "sftp --username=user --password=pass --allow-unknown-keys=true --task-launcher-output=true --remote-dir=/tmp/remote-files/ --batch-resource-uri=file:////path/to/sftp-ingest.jar --data-source-url=jdbc:h2:tcp://localhost:19092/mem:dataflow --data-source-user-name=sa --local-file-path-job-parameter-value=/tmp/local-files/ | task-launcher-local" --deploy +Created new stream 'inboundSftp' +Deployment request has been sent +---- ++ + +. Verify Stream deployment ++ +We can see the status of the streams to be deployed with `stream list`, for example: ++ +[source,console,options=nowrap] +---- +dataflow:>stream list +╔═══════════╤═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╤═════════════╗ +║Stream Name│ Stream Definition │ Status ║ +╠═══════════╪═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╪═════════════╣ +║inboundSftp│sftp --username=user --password=****** --allow-unknown-keys=true --task-launcher-output=true --remote-dir=/tmp/remote-files/ │The stream ║ +║ │--batch-resource-uri=file:///path/to/spring-cloud-dataflow-samples/batch/file-ingest-sftp/target/ingest-sftp-1.0.0.jar │has been ║ +║ │--data-source-url=jdbc:h2:tcp://localhost:19092/mem:dataflow --data-source-user-name=sa --local-file-path-job-parameter-value=/tmp/local-files/ | │successfully ║ +║ │task-launcher-local │deployed ║ +╚═══════════╧═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╧═════════════╝ +---- ++ + +. Inspecting logs ++ +In the event the stream failed to deploy, or you would like to inspect the logs for any reason, the log paths to applications created within the `inboundSftp` stream will be printed to console where the Spring Cloud Data Flow server was launched from, for example: ++ +[source,console,options=nowrap] +---- +2018-04-27 11:13:50.361 INFO 46308 --- [nio-9393-exec-8] o.s.c.d.spi.local.LocalAppDeployer : Deploying app with deploymentId inboundSftp.task-launcher-local instance 0. + Logs will be in /var/folders/6x/tgtx9xbn0x16xq2sx1j2rld80000gn/T/spring-cloud-deployer-1671726770179703111/inboundSftp-1524842030314/inboundSftp.task-launcher-local +... +... +2018-04-27 11:13:50.369 INFO 46308 --- [nio-9393-exec-8] o.s.c.d.spi.local.LocalAppDeployer : Deploying app with deploymentId inboundSftp.sftp instance 0. + Logs will be in /var/folders/6x/tgtx9xbn0x16xq2sx1j2rld80000gn/T/spring-cloud-deployer-1671726770179703111/inboundSftp-1524842030363/inboundSftp.sftp +---- ++ +In this example, the logs for the `SFTP` application would be in: ++ +``` +/var/folders/6x/tgtx9xbn0x16xq2sx1j2rld80000gn/T/spring-cloud-deployer-1671726770179703111/inboundSftp-1524842030363/inboundSftp.sftp +``` ++ +The log files contained in this directory would be useful to debug issues such as SFTP connection failures. ++ +Additionally, the logs for the `task-launcher-local` application would be in: ++ +``` +/var/folders/6x/tgtx9xbn0x16xq2sx1j2rld80000gn/T/spring-cloud-deployer-1671726770179703111/inboundSftp-1524842030314/inboundSftp.task-launcher-local +``` ++ +Since we utilize the `task-launcher-local` application to launch batch jobs upon receiving new files, this file would contain the start up logs of the `task-launcher-local` application but also print out the log paths to all applications deployed from it. The log files for each launched task can also be inspected as needed for debugging or verification. ++ + +. Add data ++ +Normally data would be arriving on an SFTP server, but since we are running this sample locally we will simulate that by adding data into the path specified by `--remote-dir`. A sample data file can be found in the `data/` directory of the sample project. ++ +Lets copy `data/people.csv` into the `/tmp/remote-files` directory which is acting as the remote SFTP server directory. This file will be detected by the SFTP application that is polling the remote directory and launch a batch job for processing. ++ +``` +$ cp data/people.csv /tmp/remote-files +``` ++ + +. Inspect Job Executions ++ +After data is received and the batch job runs, it will be recorded as a Job Execution. We can view job executions by for example issuing the following command in the Spring Cloud Data Flow shell: ++ +[source,console,options=nowrap] +---- +dataflow:>job execution list +╔═══╤═══════╤═════════╤════════════════════════════╤═════════════════════╤══════════════════╗ +║ID │Task ID│Job Name │ Start Time │Step Execution Count │Definition Status ║ +╠═══╪═══════╪═════════╪════════════════════════════╪═════════════════════╪══════════════════╣ +║1 │1 │ingestJob│Tue May 01 23:34:05 EDT 2018│1 │Destroyed ║ +╚═══╧═══════╧═════════╧════════════════════════════╧═════════════════════╧══════════════════╝ +---- ++ +As well as list more details about that specific job execution: ++ +[source,console,options=nowrap] +---- +dataflow:>job execution display --id 1 +╔═══════════════════════╤════════════════════════════╗ +║ Key │ Value ║ +╠═══════════════════════╪════════════════════════════╣ +║Job Execution Id │1 ║ +║Task Execution Id │1 ║ +║Task Instance Id │1 ║ +║Job Name │ingestJob ║ +║Create Time │Tue May 01 23:34:05 EDT 2018║ +║Start Time │Tue May 01 23:34:05 EDT 2018║ +║End Time │Tue May 01 23:34:06 EDT 2018║ +║Running │false ║ +║Stopping │false ║ +║Step Execution Count │1 ║ +║Execution Status │COMPLETED ║ +║Exit Status │COMPLETED ║ +║Exit Message │ ║ +║Definition Status │Destroyed ║ +║Job Parameters │ ║ +║run.id(LONG) │1 ║ +║remoteFilePath(STRING) │/tmp/remote-files/people.csv║ +║localFilePath(STRING) │/tmp/local-files/people.csv ║ +╚═══════════════════════╧════════════════════════════╝ +---- ++ +. Verify data ++ +When the the batch job runs, we download the file to the local directory of `/tmp/local-files` and then transform that data into uppercase names and store the data in the database. ++ +You may use any database tool that supports the H2 database to inspect the data. In this example we use the database tool `DBeaver`. Lets inspect the table to ensure our data was processed correctly. ++ +Within DBeaver, create a connection to the database using the JDBC URL of `jdbc:h2:tcp://localhost:19092/mem:dataflow`. Upon connection expand the `PUBLIC` schema, then expand `Tables` and then double click on the table `PEOPLE`. When the table data loads, click the "Data" tab and the transformed data from the CSV file will appear containing the records from the file uppercased. ++ +. Seen file caching ++ +Since we are storing file paths that have been seen on the SFTP server, updating or adding to `/tmp/remote-files/people.csv` will not cause a new batch job to run. If using the example data file above simply copy the file as a new name, for example: ++ +``` +$ cp data/people.csv /tmp/remote-files/people2.csv +``` ++ +Refreshing the contents of the database table will show the new data that was transformed and stored. The `job execution list`, `job execution display --id X` and database inspection commands above will let you view details about subsequent runs spawned from new files arriving. ++ +Alternatively you can delete a single seen files from Redis by for example: ++ +``` +127.0.0.1:6379> DEL sftpSource "/tmp/remote-files/people.csv" +(integer) 1 +127.0.0.1:6379> +``` ++ +Or delete all seen files, for example: ++ +``` +127.0.0.1:6379> DEL sftpSource +(integer) 1 +127.0.0.1:6379> +``` ++ +These files should be deleted from `/tmp/remote-files` prior to deleting them from Redis, otherwise they will be seen again and re-processed. ++ + + +==== Summary + +In this sample, you have learned: + +* How to integrate SFTP file fetching into your batch job +* How to create and launch a stream to poll files on an SFTP server and launch a batch job +* How to verify status via logs and shell commands + diff --git a/src/main/asciidoc/overview.adoc b/src/main/asciidoc/overview.adoc index 9b5094b..a4e5f11 100644 --- a/src/main/asciidoc/overview.adoc +++ b/src/main/asciidoc/overview.adoc @@ -17,6 +17,9 @@ include::streaming/custom-apps/celsius-converter-processor/main.adoc[] include::tasks/simple-batch-job/main.adoc[] include::tasks/file-ingest/main.adoc[] +== Stream Launching Batch Job +include::batch/file-ingest-sftp/main.adoc[] + == Analytics include::analytics/twitter-analytics/main.adoc[]