Chris Schaefer
committed by
Mark Pollack
Mark Pollack
parent
2960e249aa
commit
bdc9d1848d
BIN
batch/file-ingest-sftp-cf/artifacts/ingest-sftp-cf-1.0.0.jar
Normal file
BIN
batch/file-ingest-sftp-cf/artifacts/ingest-sftp-cf-1.0.0.jar
Normal file
Binary file not shown.
3
batch/file-ingest-sftp-cf/data/people.csv
Normal file
3
batch/file-ingest-sftp-cf/data/people.csv
Normal file
@@ -0,0 +1,3 @@
|
||||
Jane,Doe
|
||||
John,Doe
|
||||
Joe,Doe
|
||||
|
128
batch/file-ingest-sftp-cf/pom.xml
Normal file
128
batch/file-ingest-sftp-cf/pom.xml
Normal file
@@ -0,0 +1,128 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>ingest-sftp-cf</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>2.0.1.RELEASE</version>
|
||||
</parent>
|
||||
<properties>
|
||||
<java.version>1.8</java.version>
|
||||
<maven.compiler.plugin.version>3.7.0</maven.compiler.plugin.version>
|
||||
<spring.cloud.task.version>1.2.2.RELEASE</spring.cloud.task.version>
|
||||
<spring.integration.version>5.0.4.RELEASE</spring.integration.version>
|
||||
<commons.io.version>2.4</commons.io.version>
|
||||
<sshd.core.version>1.6.0</sshd.core.version>
|
||||
<spring.cloud.connector.version>2.0.2.RELEASE</spring.cloud.connector.version>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-batch</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.h2database</groupId>
|
||||
<artifactId>h2</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-task-core</artifactId>
|
||||
<version>${spring.cloud.task.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-task-batch</artifactId>
|
||||
<version>${spring.cloud.task.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>commons-io</groupId>
|
||||
<artifactId>commons-io</artifactId>
|
||||
<version>${commons.io.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-file</artifactId>
|
||||
<version>${spring.integration.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-sftp</artifactId>
|
||||
<version>${spring.integration.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.sshd</groupId>
|
||||
<artifactId>sshd-core</artifactId>
|
||||
<version>${sshd.core.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.hsqldb</groupId>
|
||||
<artifactId>hsqldb</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-test</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-test</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-cloudfoundry-connector</artifactId>
|
||||
<version>${spring.cloud.connector.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-spring-service-connector</artifactId>
|
||||
<version>${spring.cloud.connector.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<version>${maven.compiler.plugin.version}</version>
|
||||
<configuration>
|
||||
<source>${java.version}</source>
|
||||
<target>${java.version}</target>
|
||||
<testSource>${java.version}</testSource>
|
||||
<testTarget>${java.version}</testTarget>
|
||||
<compilerArgument>-Xlint:all</compilerArgument>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>repository.spring.milestone</id>
|
||||
<name>Spring Milestone Repository</name>
|
||||
<url>http://repo.spring.io/milestone</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
<pluginRepositories>
|
||||
<pluginRepository>
|
||||
<id>spring-milestones</id>
|
||||
<name>Spring Milestones</name>
|
||||
<url>https://repo.spring.io/milestone</url>
|
||||
<snapshots>
|
||||
<enabled>false</enabled>
|
||||
</snapshots>
|
||||
</pluginRepository>
|
||||
</pluginRepositories>
|
||||
</project>
|
||||
@@ -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,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<Person> reader(@Value("#{jobParameters['localFilePath']}") String localFilePath) throws Exception {
|
||||
return new FlatFileItemReaderBuilder<Person>()
|
||||
.name("reader")
|
||||
.resource(new FileSystemResource(localFilePath))
|
||||
.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())
|
||||
.listener(ingestStepExecutionListener(null, null))
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RemoteResource remoteResource() {
|
||||
return new SftpRemoteResource(batchConfigurationProperties.getSftpHost(), batchConfigurationProperties.getSftpPort(),
|
||||
batchConfigurationProperties.getSftpUsername(), batchConfigurationProperties.getSftpPassword());
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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,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);
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
spring.application.name=fileIngestSftp
|
||||
spring.datasource.initialization-mode=always
|
||||
@@ -0,0 +1,6 @@
|
||||
CREATE TABLE IF NOT EXISTS people (
|
||||
person_id MEDIUMINT NOT NULL AUTO_INCREMENT,
|
||||
first_name VARCHAR(20),
|
||||
last_name VARCHAR(20),
|
||||
PRIMARY KEY (person_id)
|
||||
);
|
||||
@@ -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.<NamedFactory<Command>>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<String, Object> properties = new HashMap<String, Object>();
|
||||
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<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);
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
);
|
||||
@@ -226,6 +226,282 @@ These files should be deleted from `/tmp/remote-files` prior to deleting them fr
|
||||
+
|
||||
|
||||
|
||||
==== Using the Cloud Foundry Server
|
||||
|
||||
===== Additional Prerequisites
|
||||
|
||||
* Cloud Foundry instance
|
||||
* A `mysql` service instance
|
||||
* A `rabbit` service instance
|
||||
* A `redis` service instance
|
||||
* The Spring Cloud Data Flow Cloud Foundry Server
|
||||
* An SFTP server accessible from the Cloud Foundry instance
|
||||
|
||||
The Cloud Foundry Data Flow Server is Spring Boot application available for http://cloud.spring.io/spring-cloud-dataflow/#platform-implementations/[download] or you can https://github.com/spring-cloud/spring-cloud-dataflow-server-cloudfoundry[build] it yourself.
|
||||
If you build it yourself, the executable jar will be in `spring-cloud-dataflow-server-cloudfoundry/target`
|
||||
|
||||
NOTE: Although you can run the Data Flow Cloud Foundry Server locally and configure it to deploy to any Cloud Foundry instance, we will
|
||||
deploy the server to Cloud Foundry as recommended.
|
||||
|
||||
. Verify that CF instance is reachable (Your endpoint urls will be different from what is shown here).
|
||||
+
|
||||
|
||||
```
|
||||
$ cf api
|
||||
API endpoint: https://api.system.io (API version: ...)
|
||||
|
||||
$ cf apps
|
||||
Getting apps in org [your-org] / space [your-space] as user...
|
||||
OK
|
||||
|
||||
No apps found
|
||||
```
|
||||
. Follow the instructions to deploy the https://docs.spring.io/spring-cloud-dataflow-server-cloudfoundry/docs/current/reference/htmlsingle[Spring Cloud Data Flow Cloud Foundry server]. The following manifest file can be used, replacing values as needed:
|
||||
+
|
||||
[source,console,options=nowrap]
|
||||
----
|
||||
---
|
||||
applications:
|
||||
- name: dataflow-server
|
||||
host: dataflow-server
|
||||
memory: 2G
|
||||
disk_quota: 2G
|
||||
instances: 1
|
||||
path: /PATH/TO/SPRING-CLOUD-DATAFLOW-SERVER-CLOUDFOUNDRY-JAR
|
||||
env:
|
||||
SPRING_APPLICATION_NAME: dataflow-server
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_URL: YOUR_CF_URL
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_ORG: YOUR_CF_ORG
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_SPACE: YOUR_CF_SPACE
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_DOMAIN: YOUR_CF_DOMAIN
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_USERNAME: YOUR_CF_USER
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_PASSWORD: YOUR_CF_PASSWORD
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_STREAM_SERVICES: rabbit
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_TASK_SERVICES: mysql
|
||||
SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_SKIP_SSL_VALIDATION: true
|
||||
SPRING_APPLICATION_JSON: '{"maven": { "remote-repositories": { "repo1": { "url": "https://repo.spring.io/libs-release"}, "repo2": { "url": "https://repo.spring.io/libs-snapshot"}, "repo3": { "url": "https://repo.spring.io/libs-milestone"} } } }'
|
||||
services:
|
||||
- mysql
|
||||
- redis
|
||||
----
|
||||
+
|
||||
If your Cloud Foundry installation is behind a firewall, you may need to install the stream apps used in this sample in your internal Maven repository and https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-maven[configure] the server to access that repository.
|
||||
. Once you have successfully executed `cf push`, verify the dataflow server is running
|
||||
+
|
||||
|
||||
```
|
||||
$ cf apps
|
||||
Getting apps in org [your-org] / space [your-space] as user...
|
||||
OK
|
||||
|
||||
name requested state instances memory disk urls
|
||||
dataflow-server started 1/1 1G 1G dataflow-server.app.io
|
||||
```
|
||||
|
||||
. Notice that the `dataflow-server` application is started and ready for interaction via the url endpoint
|
||||
|
||||
. Connect the `shell` with `server` running on Cloud Foundry, e.g., `http://dataflow-server.app.io`
|
||||
+
|
||||
```
|
||||
$ cd <PATH/TO/SPRING-CLOUD-DATAFLOW-SHELL-JAR>
|
||||
$ java -jar spring-cloud-dataflow-shell-<VERSION>.jar
|
||||
|
||||
____ ____ _ __
|
||||
/ ___| _ __ _ __(_)_ __ __ _ / ___| | ___ _ _ __| |
|
||||
\___ \| '_ \| '__| | '_ \ / _` | | | | |/ _ \| | | |/ _` |
|
||||
___) | |_) | | | | | | | (_| | | |___| | (_) | |_| | (_| |
|
||||
|____/| .__/|_| |_|_| |_|\__, | \____|_|\___/ \__,_|\__,_|
|
||||
____ |_| _ __|___/ __________
|
||||
| _ \ __ _| |_ __ _ | ___| | _____ __ \ \ \ \ \ \
|
||||
| | | |/ _` | __/ _` | | |_ | |/ _ \ \ /\ / / \ \ \ \ \ \
|
||||
| |_| | (_| | || (_| | | _| | | (_) \ V V / / / / / / /
|
||||
|____/ \__,_|\__\__,_| |_| |_|\___/ \_/\_/ /_/_/_/_/_/
|
||||
|
||||
|
||||
Welcome to the Spring Cloud Data Flow shell. For assistance hit TAB or type "help".
|
||||
server-unknown:>
|
||||
```
|
||||
+
|
||||
```
|
||||
server-unknown:>dataflow config server http://dataflow-server.app.io
|
||||
Successfully targeted http://dataflow-server.app.io
|
||||
dataflow:>
|
||||
```
|
||||
+
|
||||
|
||||
|
||||
===== Building and Running the Demo
|
||||
|
||||
. Build the demo JAR
|
||||
+
|
||||
Building upon the code in `batch/file-ingest-sftp`, in this demo we utilize https://cloud.spring.io/spring-cloud-connectors/[Spring Cloud Connectors] to automatically bind Cloud Foundry services such as MySQL and Redis.
|
||||
+
|
||||
From the root of this project:
|
||||
+
|
||||
```
|
||||
$ cd batch/file-ingest-sftp-cf
|
||||
$ mvn clean package
|
||||
```
|
||||
+
|
||||
The resulting `target/ingest-sftp-cf-1.0.0.jar` artifact must be uploaded to a remote location such as an HTTP server or Maven repository that is accessible to your Cloud Foundry installation. For convenience, a pre-built demo artifact can be found at: https://raw.githubusercontent.com/spring-cloud/spring-cloud-dataflow-samples/master/batch/file-ingest-sftp-cf/artifacts/ingest-sftp-cf-1.0.0.jar[https://raw.githubusercontent.com/spring-cloud/spring-cloud-dataflow-samples/master/batch/file-ingest-sftp-cf/artifacts/ingest-sftp-cf-1.0.0.jar]
|
||||
|
||||
. Create the data directory
|
||||
+
|
||||
A directory must be created on the SFTP server where the batch job will find files and download for processing. This path must exist prior to running the batch job can can be any location that is accessible by the configured SFTP user. On the SFTP server create a directory, for example:
|
||||
+
|
||||
```
|
||||
$ mkdir /tmp/remote-files
|
||||
```
|
||||
+
|
||||
|
||||
. Register the the SFTP source and the Task Launcher Cloud Foundry sink
|
||||
+
|
||||
With the Spring Cloud Data Flow server running, the `SFTP` source and `task-launcher-cloudfoundry` sink needs to be registered. The `SFTP` source application will do the work of polling for new files and when received, it sends a message to the `task-launcher-cloudfoundry` 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-rabbit:2.0.0.BUILD-SNAPSHOT
|
||||
Successfully registered application 'source:sftp'
|
||||
dataflow:>app register --name task-launcher-cloudfoundry --type sink --uri maven://org.springframework.cloud.stream.app:task-launcher-cloudfoundry-sink-rabbit:2.0.0.BUILD-SNAPSHOT
|
||||
Successfully registered application 'sink:task-launcher-local'
|
||||
----
|
||||
+
|
||||
|
||||
. Create and deploy the stream
|
||||
+
|
||||
Now a stream needs to be created that will poll the SFTP server, launching the batch job when new files arrive.
|
||||
+
|
||||
Create the stream:
|
||||
+
|
||||
NOTE: You must replace `--username=user`, `--password=pass` and `--host=1.1.1.1` below to their respective values. The `--username=` and `--password=` parameters are the credentials for your remote SFTP user. The `--batch-resource-uri=` parameter is the path to the batch artifact to use. In this Stream definition, the published sample batch artifact JAR is used. If you would like to use a custom built artifact, replace this value with the artifact location.
|
||||
+
|
||||
[source,console]
|
||||
----
|
||||
dataflow:>stream create --name inboundSftp --definition "sftp --username=user --password=pass --host=1.1.1.1 --allow-unknown-keys=true --task-launcher-output=true --remote-dir=/tmp/remote-files --batch-resource-uri=https://raw.githubusercontent.com/spring-cloud/spring-cloud-dataflow-samples/master/batch/file-ingest-sftp-cf/artifacts/ingest-sftp-cf-1.0.0.jar --local-file-path-job-parameter-value=/tmp/ | task-launcher-cloudfoundry --spring.cloud.deployer.cloudfoundry.services=mysql"
|
||||
Created new stream 'inboundSftp'
|
||||
dataflow:>
|
||||
----
|
||||
+
|
||||
Deploy the stream:
|
||||
+
|
||||
NOTE: You must replace `CF_USER`, `CF_PASSWORD`, `CF_ORG`, `CF_SPACE`, and `CF_URL` below with the appropriate values for your setup. The values will be used by the task launcher to launch tasks.
|
||||
+
|
||||
[source,console]
|
||||
----
|
||||
dataflow:>stream deploy inboundSftp --properties "app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.username=CF_USER,app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.password=CF_PASSWORD,app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.org=CF_ORG,app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.space=CF_SPACE,app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.url=CF_URL,app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.skip-ssl-validation=true,app.task-launcher-cloudfoundry.spring.cloud.deployer.cloudfoundry.apiTimeout=30000,deployer.sftp.cloudfoundry.services=redis"
|
||||
Deployment request has been sent for stream 'inboundSftp'
|
||||
|
||||
dataflow:>
|
||||
----
|
||||
+
|
||||
|
||||
. Verify Stream deployment
|
||||
+
|
||||
The status of the stream to be deployed can be queried with `stream list`, for example:
|
||||
+
|
||||
[source,console,options=nowrap]
|
||||
----
|
||||
dataflow:>stream list
|
||||
╔═══════════╤═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
|
||||
║Stream Name│ Stream Definition │ Status ║
|
||||
╠═══════════╪═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╣
|
||||
║inboundSftp│sftp --password='******' --local-file-path-job-parameter-value=/tmp/ --host=1.1.1.1 --remote-dir=/tmp/remote-files --allow-unknown-keys=true │The stream has ║
|
||||
║ │--task-launcher-output=true |been successfully ║
|
||||
║ |--batch-resource-uri=https://raw.githubusercontent.com/spring-cloud/spring-cloud-dataflow-samples/master/batch/file-ingest-sftp-cf/artifacts/ingest-sftp-cf-1.0.0.jar |deployed ║
|
||||
║ |--username=user | task-launcher-cloudfoundry --spring.cloud.deployer.cloudfoundry.services=mysql | ║
|
||||
╚═══════════╧═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
|
||||
----
|
||||
+
|
||||
|
||||
. Inspecting logs
|
||||
+
|
||||
In the event the stream failed to deploy, or you would like to inspect the logs for any reason, the logs can be obtained from individual applications. First list the deployed apps:
|
||||
+
|
||||
[source,console,options=nowrap]
|
||||
----
|
||||
$ cf apps
|
||||
Getting apps in org cf_org / space cf_space as cf_user...
|
||||
OK
|
||||
|
||||
name requested state instances memory disk urls
|
||||
dataflow-server started 1/1 2G 2G dataflow-server.app.io
|
||||
dataflow-server-N5RYLDj-inboundSftp-sftp started 1/1 1G 1G dataflow-server-N5RYLDj-inboundSftp-sftp.dataflow-server.app.io
|
||||
dataflow-server-N5RYLDj-inboundSftp-task-launcher-cloudfoundry started 1/1 1G 1G dataflow-server-N5RYLDj-inboundSftp-task-launcher-cloudfoundry.dataflow-server.app.io
|
||||
----
|
||||
+
|
||||
In this example, the logs for the `SFTP` application can be viewed by:
|
||||
+
|
||||
```
|
||||
cf logs dataflow-server-N5RYLDj-inboundSftp-sftp --recent
|
||||
```
|
||||
+
|
||||
The log files of this application would be useful to debug issues such as SFTP connection failures.
|
||||
+
|
||||
Additionally, the logs for the `task-launcher-local` application can be viewed by:
|
||||
+
|
||||
```
|
||||
cf logs dataflow-server-N5RYLDj-inboundSftp-task-launcher-cloudfoundry --recent
|
||||
```
|
||||
+
|
||||
Since the `task-launcher-cloudfoundry` application is used to launch batch jobs upon receiving new files, this log would contain the start up logs of the `task-launcher-cloudfoundry` application but also log the name and other information of all applications deployed from it. The application log file for each launched task can also be inspected as needed for debugging or verification.
|
||||
+
|
||||
|
||||
. Add data
|
||||
+
|
||||
A sample data file can be found in the `data/` directory of the sample project. Copy `data/people.csv` into the `/tmp/remote-files` directory of 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.
|
||||
+
|
||||
|
||||
. 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│Thu Jun 07 13:46:42 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 │Thu Jun 07 13:46:42 EDT 2018║
|
||||
║Start Time │Thu Jun 07 13:46:42 EDT 2018║
|
||||
║End Time │Thu Jun 07 13:46:44 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/1012.csv ║
|
||||
║localFilePath(STRING) │/tmp/1012.csv ║
|
||||
╚═══════════════════════╧════════════════════════════╝
|
||||
----
|
||||
+
|
||||
. Verification of Data and Seen Files
|
||||
+
|
||||
Verification of data loaded by the batch job and seen file tracking can be accomplished in the same way as with Local Server using the appropriate tools. Consult the documentation for the service broker on your platform (PWS, PCF, etc) for information on how to connect to the backing service.
|
||||
+
|
||||
|
||||
|
||||
==== Summary
|
||||
|
||||
In this sample, you have learned:
|
||||
@@ -233,4 +509,5 @@ 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
|
||||
* How to run the SFTP file ingest batch job on Cloud Foundry
|
||||
|
||||
|
||||
Reference in New Issue
Block a user