SpringCloud/spring-cloud-dataflow-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Main File Ingest Demo written

Browse Source 
* Add Concurrent Execution Limit
* Add Avoid Duplicate Processing
* Update link
* Add snapshot repo
* Update copyright year
* Finish Metadata Store sample and reduce number of split files to 20
* Add Cloud Foundry MetadataStore
This commit is contained in:
David Turanski
2018-10-26 15:09:59 -04:00
committed by Chris Schaefer
parent 4a8fbbd723
commit 699f7856e4
72 changed files with 11783 additions and 2215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

211
src/main/asciidoc/batch/file-ingest-sftp/local.adoc Normal file
View File

@@ -0,0 +1,211 @@
[[sftp-file-ingest-local]]
==== Using the Local Server
===== Additional Prerequisites
* A running local Data Flow Server
include::{docs_dir}/local-server.adoc[]
* Running instance of link:http://kafka.apache.org/downloads.html[Kafka]
* 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.
===== Building and Running the Demo
. Build the demo JAR
+
From the root of this project:
+
```
$ cd batch/file-ingest
$ mvn clean package
```
NOTE: For convenience, you can skip this step.
The jar is published to the https://repo.spring.io/libs-snapshot-local/io/spring/cloud/dataflow/ingest/ingest/1.0.0.BUILD-SNAPSHOT/[Spring Maven repository]
+
. Create the data directories
+
Now we create a remote directory on the SFTP server and a local directory where the batch job expects to find files.
+
NOTE: If you are using a remote SFTP server, create the remote directory on the SFTP server.
Since we are using the local machine as the SFTP server, we will create both the local and remote directories on the local machine.
+
```
$ mkdir -p /tmp/remote-files /tmp/local-files
```
+
. Register the `sftp-dataflow` source and the `task-launcher-dataflow` sink
+
With our Spring Cloud Data Flow server running, we register the `sftp-dataflow` source and `task-launcher-dataflow` sink.
The `sftp-dataflow` source application will do the work of polling the remote directory for new files and downloading them to the local directory.
As each file is received, it emits a message for the `task-launcher-dataflow` sink to launch the task to process the data from 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-dataflow-source-kafka:2.0.3.BUILD-SNAPSHOT
Successfully registered application 'source:sftp'
dataflow:>app register --name task-launcher --type sink --uri maven://org.springframework.cloud.stream.app:task-launcher-dataflow-sink-kafka:1.0.0.BUILD-SNAPSHOT
Successfully registered application 'sink:task-launcher'
----
+
. Register and create the file ingest task. If you're using the published jar, set `--uri maven://io.spring.cloud.dataflow.ingest:ingest:1.0.0.BUILD-SNAPSHOT`:
[source,console,options=nowrap]
dataflow:>app register --name fileIngest --type task --uri file:///path/to/target/ingest-X.X.X.jar
Successfully registered application 'task:fileIngest'
dataflow:>task create fileIngestTask --definition fileIngest
Created new task 'fileIngestTask'
+
. Create and deploy the stream
+
Now lets create and deploy the stream.
Once deployed, the stream will start polling the SFTP server and, when new files arrive, launch the batch job.
+
NOTE: Replace `<user>` and '<pass>` below.
The `<username>` and `<password>` values are the credentials for the local (or remote) user.
If not using a local SFTP server, specify the host using the `--host`, and optionally `--port`, parameters.
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.launch.request.taskName=fileIngestTask --remote-dir=/tmp/remote-files/ --local-dir=/tmp/local-files/ | task-launcher" --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 --password='******' --remote-dir=/tmp/remote-files/ --local-dir=/tmp/local-files/ --task.launch.request.taskName=fileIngestTask│The stream has been ║
║ │--allow-unknown-keys=true --username=<user> | task-launcher │successfully deployed ║
╚═══════════╧════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╧════════════════════════════╝
----
+
. Inspect logs
+
In the event the stream failed to deploy, or you would like to inspect the logs for any reason, you can get the location of the logs to applications created for the `inboundSftp` stream using the `runtime apps` command:
[source,console,options=nowrap]
dataflow:>runtime apps
╔═══════════════════════════╤═══════════╤════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║ App Id / Instance Id │Unit Status│ No. of Instances / Attributes ║
╠═══════════════════════════╪═══════════╪════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╣
║inboundSftp.sftp │ deployed │ 1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 23057 ║
║ │ │ pid = 71927 ║
║ │ │ port = 23057 ║
║inboundSftp.sftp-0 │ deployed │ stderr = /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-deployer-120915912946760306/inboundSftp-1540821009913/inboundSftp.sftp/stderr_0.log ║
║ │ │ stdout = /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-deployer-120915912946760306/inboundSftp-1540821009913/inboundSftp.sftp/stdout_0.log ║
║ │ │ url = http://192.168.64.1:23057 ║
║ │ │working.dir = /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-deployer-120915912946760306/inboundSftp-1540821009913/inboundSftp.sftp ║
╟───────────────────────────┼───────────┼────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╢
║inboundSftp.task-launcher │ deployed │ 1 ║
╟┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┼┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈╢
║ │ │ guid = 60081 ║
║ │ │ pid = 71926 ║
║ │ │ port = 60081 ║
║inboundSftp.task-launcher-0│ deployed │ stderr = /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-deployer-120915912946760306/inboundSftp-1540820991695/inboundSftp.task-launcher/stderr_0.log║
║ │ │ stdout = /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-deployer-120915912946760306/inboundSftp-1540820991695/inboundSftp.task-launcher/stdout_0.log║
║ │ │ url = http://192.168.64.1:60081 ║
║ │ │working.dir = /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-deployer-120915912946760306/inboundSftp-1540820991695/inboundSftp.task-launcher ║
╚═══════════════════════════╧═══════════╧════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
+
. Add data
+
Normally data would be uploaded to an SFTP server.
We will simulate this by copying a file into the directory specified by `--remote-dir`.
Sample data can be found in the `data/` directory of the <<Batch File Ingest>> project.
+
Copy `data/name-list.csv` into the `/tmp/remote-files` directory which the SFTP source is monitoring.
When this file is detected, the `sftp` source will download it to the `/tmp/local-files` directory specified by `--local-dir`, and emit a Task Launch Request.
The Task Launch Request includes the name of the task to launch along with the local file path, given as the command line argument `localFilePath`.
Spring Batch binds each command line argument to a corresponding JobParameter.
The FileIngestTask job processes the file given by the JobParameter named `localFilePath`.
The `task-launcher` sink polls for messages using an exponential back-off.
Since there have not been any recent requests, the task will launch within 30 seconds after the request is published.
+
```
$ cp data/name-list.csv /tmp/remote-files
```
When the batch job launches, you will see something like this in the SCDF console log:
[source,console,options=nowrap]
2018-10-26 16:47:24.879 INFO 86034 --- [nio-9393-exec-7] o.s.c.d.spi.local.LocalTaskLauncher : Command to be executed: /Library/Java/JavaVirtualMachines/jdk1.8.0_60.jdk/Contents/Home/jre/bin/java -jar <path-to>/batch/file-ingest/target/ingest-1.0.0.jar localFilePath=/tmp/local-files/name-list.csv --spring.cloud.task.executionid=1
2018-10-26 16:47:25.100 INFO 86034 --- [nio-9393-exec-7] o.s.c.d.spi.local.LocalTaskLauncher : launching task fileIngestTask-8852d94d-9dd8-4760-b0e4-90f75ee028de
Logs will be in /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/fileIngestTask3100511340216074735/1540586844871/fileIngestTask-8852d94d-9dd8-4760-b0e4-90f75ee028de
+
. 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 │Created ║
╚═══╧═══════╧═════════╧════════════════════════════╧═════════════════════╧══════════════════╝
----
+
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 │Fri Oct 26 16:57:51 EDT 2018 ║
║Start Time │Fri Oct 26 16:57:51 EDT 2018 ║
║End Time │Fri Oct 26 16:57:53 EDT 2018 ║
║Running │false ║
║Stopping │false ║
║Step Execution Count │1 ║
║Execution Status │COMPLETED ║
║Exit Status │COMPLETED ║
║Exit Message │ ║
║Definition Status │Created ║
║Job Parameters │ ║
║-spring.cloud.task.executionid(STRING) │1 ║
║run.id(LONG) │1 ║
║localFilePath(STRING) │/tmp/local-files/name-list.csv║
╚═══════════════════════════════════════╧══════════════════════════════╝
----
+
. Verify data
+
When the the batch job runs, it processes the file in the local directory `/tmp/local-files` and transforms each item to uppercase names and inserts it into 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 `jdbc:h2:tcp://localhost:19092/mem:dataflow`, and user `sa` with no password.
When connected, expand the `PUBLIC` schema, then expand `Tables` and then double click on the table `PEOPLE`.
When the table data loads, click the "Data" tab to view the data.
+
. You're done!

613
src/main/asciidoc/batch/file-ingest-sftp/main.adoc
View File

@@ -1,513 +1,246 @@
[[spring-cloud-data-flow-samples-sftp-file-ingest-overview]]
:sectnums:
:docs_dir: ../..
=== Batch File Ingest - SFTP Demo
=== Batch File Ingest - SFTP
In the <<Batch File Ingest>> sample we built a link:https://projects.spring.io/spring-batch[Spring Batch] application that link:https://cloud.spring.io/spring-cloud-dataflow[Spring Cloud Data Flow] launched as a task to process a file.
This time we will build on that sample to create and deploy a link:https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#spring-cloud-dataflow-streams[stream] that launches that task.
The stream will poll an SFTP server and, for each new file that it finds, will download the file and launch the batch job to process it.
In the <<Batch File Ingest>> 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.
The source for the demo project is located in the `batch/file-ingest` directory at the top-level of this repository.
==== 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.
include::local.adoc[]
* A database tool such as link:https://dbeaver.jkiss.org/download/[DBeaver] to inspect the database contents
include::pcf.adoc[]
NOTE: To simplify the dependencies and configuration in this example, we will use our local machine acting as an SFTP server.
==== Limiting Concurrent Task Executions
==== Batch File Ingest SFTP Demo Overview
The Batch File Ingest - SFTP Demo processes a single file with 5000+ items. What if we copy 100 files to the remote directory?
The sftp source will process them immediately, generating 100 task launch requests. The Dataflow Server launches tasks asynchronously so this could potentially overwhelm the resources of the runtime platform.
For example, when running the Data Flow server on your local machine, each launched task creates a new JVM. In Cloud Foundry, each task creates a new container instance.
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 <<Batch File Ingest>>.
Fortunately, Spring Cloud Data Flow 1.7 introduced new features to manage concurrently running tasks, including a new configuration parameter, `spring.cloud.dataflow.task.maximum-concurrent-tasks`, to http://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#spring-cloud-dataflow-task-limit-concurrent-executions[limit the number of concurrently running tasks].
We can use this demo to see how this works.
The key modifications from the <<Batch File Ingest>> sample are:
===== Configuring the SCDF server
Set the maximum concurrent tasks to 3.
For the local server, restart the server, adding a command line argument `--spring.cloud.dataflow.task.maximum-concurrent-tasks=3`.
* `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`.
For the Cloud Foundry server, `cf set-env <dataflow-server> SPRING_CLOUD_DATAFLOW_TASK_MAXIMUM_CONCURRENT_TASKS 3`, and restage.
* `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.
===== Running the demo
Follow the main demo instructions but change the `Add Data` step, as described below.
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
. Monitor the task launcher
+
From the root of this project:
Tail the logs on the `task-launcher` app.
+
If there are no requests in the input queue, you will see something like:
+
[source, console, options=nowrap]
----
07:42:51.760 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received- increasing polling period to 2 seconds.
07:42:53.768 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received- increasing polling period to 4 seconds.
07:42:57.780 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received- increasing polling period to 8 seconds.
07:43:05.791 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received- increasing polling period to 16 seconds.
07:43:21.801 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received- increasing polling period to 30 seconds.
07:43:51.811 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received
07:44:21.824 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received
07:44:51.834 INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : No task launch request received
----
+
The first three messages show the exponential backoff at start up or after processing the final request.
The the last three message show the task launcher in a steady state of polling for messages every 30 seconds.
Of course, these values are configurable.
+
The task launcher sink polls the input destination. The polling period adjusts according to the presence of task launch requests and also to the number of currently running tasks reported via the Data Flow server's `tasks/executions/current` REST endpoint.
The sink queries this endpoint and will pause polling the input for new requests if the number of concurrent tasks is at its limit.
This introduces a 1-30 second lag between the creation of the task launch request and the execution of the request, sacrificing some performance for resilience.
Task launch requests will never be sent to a dead letter queue because the server is busy or unavailable.
The exponential backoff also prevents the app from querying the server excessively when there are no task launch requests.
+
You can also monitor the Data Flow server:
+
[source, console, options=nowrap]
----
$ watch curl <dataflow-server-url>/tasks/executions/current
Every 2.0s: curl http://localhost:9393/tasks/executions/current ultrafox.local: Wed Oct 31 08:38:53 2018
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
0 0 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 0100 53 0 53 0 0 53 0 --:--:-- --:--:-- --:--:-- 5888
{"maximumTaskExecutions":3,"runningExecutionCount":0}
----
. Add Data
+
The directory `batch/file-ingest/data/split` contains the contents of
`batch/file-ingest/data/name-list.csv` split into 20 files, not 100 but enough to illustrate the concept.
Upload these files to the SFTP remote directory, e.g.,
```
$ cd batch/file-ingest-sftp
$ mvn clean package
sftp>cd remote-files
sftp>lcd batch/file-ingest/data/split
sftp>mput *
```
+
. 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.
+
Or if using the local machine as the SFTP server:
```
$ mkdir -p /tmp/remote-files /tmp/local-files
>cp * /tmp/remote-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]
In the `task-launcher` logs, you should now see:
[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
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling period reset to 1000 ms.
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Launching Task fileIngestTask
WARN o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Data Flow server has reached its concurrent task execution limit: (3)
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling paused- increasing polling period to 2 seconds.
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling resumed
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Launching Task fileIngestTask
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling period reset to 1000 ms.
WARN o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Data Flow server has reached its concurrent task execution limit: (3)
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling paused- increasing polling period to 2 seconds.
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling resumed
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Launching Task fileIngestTask
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling period reset to 1000 ms.
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Launching Task fileIngestTask
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Launching Task fileIngestTask
WARN o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Data Flow server has reached its concurrent task execution limit: (3)
INFO o.s.c.s.a.t.l.d.s.LaunchRequestConsumer : Polling paused- increasing polling period to 2 seconds.
...
...
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
----
==== Avoid Duplicate Processing
The `sftp` source will not process files that it has already seen.
It uses a https://docs.spring.io/spring-integration/docs/current/reference/html/system-management-chapter.html#metadata-store[Metadata Store] to keep track of files by extracting content from messages at runtime.
Out of the box, it uses an in-memory Metadata Store.
Thus, if we re-deploy the stream, this state is lost and files will be reprocessed.
Thanks to the magic of Spring, we can inject one of the available persistent Metadata Stores.
In this example, we will use the https://github.com/spring-cloud-stream-app-starters/core/tree/master/common/app-starters-metadata-store-common#jdbc[JDBC Metadata Store] since we are already using a database.
. Configure and Build the SFTP source
+
In this example, the logs for the `SFTP` application would be in:
For this we add some JDBC dependencies to the `sftp-dataflow` source.
+
Clone the https://github.com/spring-cloud-stream-app-starters/sftp[sftp] stream app starter.
From the sftp directory. Replace <binder> below with `kafka` or `rabbit` as appropriate for your configuration:
+
```
/var/folders/6x/tgtx9xbn0x16xq2sx1j2rld80000gn/T/spring-cloud-deployer-1671726770179703111/inboundSftp-1524842030363/inboundSftp.sftp
$ ./mvnw clean install -DskipTests -PgenerateApps
$ cd apps/sftp-dataflow-source-<binder>
```
+
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:
Add the following dependencies to `pom.xml`:
+
```
/var/folders/6x/tgtx9xbn0x16xq2sx1j2rld80000gn/T/spring-cloud-deployer-1671726770179703111/inboundSftp-1524842030314/inboundSftp.task-launcher-local
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jdbc</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
</dependency>
```
+
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.
If you are running on a local server with the in memory H2 database, set the JDBC url in `src/main/resources/application.properties` to use the Data Flow server's database:
+
```
$ cp data/people.csv /tmp/remote-files
spring.datasource.url=jdbc:h2:tcp://localhost:19092/mem:dataflow
```
+
. 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:
If you are running in Cloud Foundry, we will bind the source to the `mysql` service. Add the following property to `src/main/resources/application.properties`:
+
```
$ cp data/people.csv /tmp/remote-files/people2.csv
spring.integration.jdbc.initialize-schema=always
```
+
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:
Build the app:
+
```
127.0.0.1:6379> DEL sftpSource "/tmp/remote-files/people.csv"
(integer) 1
127.0.0.1:6379>
$./mvnw clean package
```
. Register the jar
+
Or delete all seen files, for example:
If running in Cloud Foundry, the resulting executable jar file must be available in a location that is accessible to your Cloud Foundry instance, such as an HTTP server or Maven repository.
If running on a local server:
+
```
127.0.0.1:6379> DEL sftpSource
(integer) 1
127.0.0.1:6379>
dataflow>app register --name sftp --type source --uri file:<project-directory>/sftp/apps/sftp-dataflow-source-kafka/target/sftp-dataflow-source-kafka-X.X.X.jar --force
```
. Run the Demo
+
These files should be deleted from `/tmp/remote-files` prior to deleting them from Redis, otherwise they will be seen again and re-processed.
Follow the instructions for building and running the main SFTP File Ingest demo, for your preferred platform, up to the `Add Data Step`.
If you have already completed the main exercise, restore the data to its initial state, and redeploy the stream:
+
==== 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).
* Clean the data directories (e.g., `tmp/local-files` and `tmp/remote-files`)
* Execute the SQL command `DROP TABLE PEOPLE;` in the database
* Undeploy the stream, and deploy it again to run the updated `sftp` source
+
```
$ 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`
If you are running in Cloud Foundry, set the deployment properties to bind `sftp` to the `mysql` service. For example:
+
```
$ 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:>
dataflow>stream deploy inboundSftp --properties "deployer.sftp.cloudfoundry.services=nfs,mysql"
```
. Add Data
+
Let's use one small file for this.
The directory `batch/file-ingest/data/split` contains the contents of
`batch/file-ingest/data/name-list.csv` split into 20 files. Upload one of them:
+
```
server-unknown:>dataflow config server http://dataflow-server.app.io
Successfully targeted http://dataflow-server.app.io
dataflow:>
sftp>cd remote-files
sftp>lcd batch/file-ingest/data/split
sftp>put names_aa.csv
```
+
===== 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:
Or if using the local machine as the SFTP server:
+
```
$ cd batch/file-ingest-sftp-cf
$ mvn clean package
$cp names_aa.csv truncate INT_METADATA_STORE;
```
. Inspect data
+
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
Using a Database browser, as described in the main demo, view the contents of the `INT_METADATA_STORE` table.
+
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:
image::metadata_store_1.png[title="JDBC Metadata Store"]
+
Note that there is a single key-value pair, where the key identies the file name (the prefix `sftpSource/` provides a namespace for the `sftp` source app) and the value is a timestamp indicating when the message was received.
The metadata store tracks files that have already been processed.
This prevents the same files from being pulled every from the remote directory on every polling cycle. Only new files, or files that have been updated will be processed.
Since there are no uniqueness constraints on the data, a file processed multiple times by our batch job will result in duplicate entries.
+
If we view the `PEOPLE` table, it should look something like this:
+
image::people_table_1.png[title="People Data"]
+
Now let's update the remote file, using SFTP `put` or if using the local machine as an SFTP server:
+
```
$ mkdir /tmp/remote-files
$touch /tmp/remote-files/names_aa.csv
```
Now the `PEOPLE` table will have duplicate data. If you `ORDER BY FIRST_NAME`, you will see something like this:
+
. Register the the SFTP source and the Task Launcher Cloud Foundry sink
image::people_table_2.png[title="People Data with Duplicates"]
+
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.
+
Of course, if we drop another one of files into the remote directory, that will processed and we will see another entry in the Metadata Store.
==== 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
* How to run the SFTP file ingest batch job on Cloud Foundry
* How to process SFTP files with a batch job
* How to create a stream to poll files on an SFTP server and launch a batch job
* How to verify job status via logs and shell commands
* How the Data Flow Task Launcher limits concurrent task executions
* How to avoid duplicate processing of files

253
src/main/asciidoc/batch/file-ingest-sftp/pcf.adoc Normal file
View File

@@ -0,0 +1,253 @@
[sftp-file-ingest-local]]
==== Using the Cloud Foundry Server
===== Additional Prerequisites
NOTE: Running this demo in Cloud Foundry requires a shared file system that is accessed by apps running in different containers.
This feature is provided by https://docs.pivotal.io/pivotalcf/2-3/devguide/services/using-vol-services.html[NFS Volume Services].
To use Volume Services with SCDF, it is required that we provide `nfs` configuration via `cf create-service` rather than `cf bind-service`.
Cloud Foundry introduced the `cf create-service` configuration option for Volume Services in version 2.3.
* A Cloud Foundry instance v2.3+ with NFS Volume Services https://docs.pivotal.io/pivotalcf/2-3/opsguide/enable-vol-services.html[enabled]
* An SFTP server accessible from the Cloud Foundry instance
* An `nfs` service instance properly configured
NOTE: For this example, we use an NFS host configured to allow https://www.tldp.org/HOWTO/NFS-HOWTO/server.html[read-write access] to the Cloud Foundry instance.
Create the `nfs` service instance using a command as below, where `share` specifies the NFS host and shared directory(`/export`), `uid` an `gid` specify an account that has read-write access to the shared directory, and `mount` is the container's mount path for each application bound to `nfs`:
```
$ cf create-service nfs Existing nfs -c '{"share":"<nfs_host_ip>/export","uid":"<uid>","gid":"<gid>", "mount":"/var/scdf"}'
```
* A `mysql` service instance
* A `rabbit` service instance
* https://github.com/pivotal-cf/PivotalMySQLWeb[PivotalMySQLWeb] or another database tool to view the data
* The Spring Cloud Data Flow Cloud Foundry Server
include::{docs_dir}/cloudfoundry-server.adoc[]
===== Configuring the SCDF server
For convenience, we will configure the SCDF server to bind all stream and task apps to the `nfs` service. Using the Cloud Foundry CLI,
set the following environment variables (or set them in the manifest):
```
cf set-env <dataflow-server-app-name> SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_STREAM_SERVICES rabbitmq,nfs
cf set-env <dataflow-server-app-name> SPRING_CLOUD_DEPLOYER_CLOUDFOUNDRY_TASK_SERVICES mysql,nfs
```
NOTE: Normally, for security and operational efficiency, we may want more fine grained control of which apps bind to the nfs service.
One way to do this is to set deployment properties when creating and deploying the stream, as shown below.
===== Running the Demo
The source code for the <<Batch File Ingest>> batch job is located in `batch/file-ingest`.
The resulting executable jar file must be available in a location that is accessible to your Cloud Foundry instance, such as an HTTP server or Maven repository.
For convenience, the jar is published to the https://repo.spring.io/libs-snapshot-local/io/spring/cloud/dataflow/ingest/ingest/1.0.0.BUILD-SNAPSHOT/[Spring Maven repository]
. Create the remote directory
+
Create a directory on the SFTP server where the `sftp` source will detect files and download them for processing.
This path must exist prior to running the demo and can be any location that is accessible by the configured SFTP user.
On the SFTP server create a directory called `remote-files`, for example:
+
```
sftp> mkdir remote-files
```
+
. Create a shared NFS directory
+
Create a directory on the NFS server that is accessible to the user, specified by `uid` and `gid`, used to create the nfs service:
+
```
$ sudo mkdir /export/shared-files
$ sudo chown <uid>:<gid> /export/shared-files
```
. Register the `sftp-dataflow` source and the `tasklauncher-dataflow` sink
+
With our Spring Cloud Data Flow server running, we register the `sftp-dataflow` source and `task-launcher-dataflow` sink.
The `sftp-dataflow` source application will do the work of polling the remote directory for new files and downloading them to the local directory.
As each file is received, it emits a message for the `task-launcher-dataflow` sink to launch the task to process the data from 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-dataflow-source-rabbit:2.0.3.BUILD-SNAPSHOT
Successfully registered application 'source:sftp'
dataflow:>app register --name task-launcher --type sink --uri maven://org.springframework.cloud.stream.app:task-launcher-dataflow-sink-rabbit:1.0.0.BUILD-SNAPSHOT
Successfully registered application 'sink:task-launcher'
----
+
. Register and create the file ingest task:
[source,console,options=nowrap]
dataflow:>app register --name fileIngest --type task --uri maven://io.spring.cloud.dataflow.ingest:ingest:1.0.0.BUILD-SNAPSHOT
Successfully registered application 'task:fileIngest'
dataflow:>task create fileIngestTask --definition fileIngest
Created new task 'fileIngestTask'
+
. Create and deploy the stream
+
Now lets create and deploy the stream.
Once deployed, the stream will start polling the SFTP server and, when new files arrive, launch the batch job.
+
NOTE: Replace `<user>`, '<pass>`, and `<host>` below.
The `<host>` is the SFTP server host, `<user>` and `<password>` values are the credentials for the remote user.
Additionally, replace `--spring.cloud.dataflow.client.server-uri=http://<dataflow-server-route>` with the URL of your dataflow server, as shown by `cf apps`.
If you have security enabled for the SCDF server, set the appropriate `spring.cloud.dataflow.client` options.
+
[source, console, options=nowrap]
----
dataflow:> app info --name task-launcher --type sink
╔══════════════════════════════╤══════════════════════════════╤══════════════════════════════╤══════════════════════════════╗
║ Option Name │ Description │ Default │ Type ║
╠══════════════════════════════╪══════════════════════════════╪══════════════════════════════╪══════════════════════════════╣
║spring.cloud.dataflow.client.a│The login username. │<none> │java.lang.String ║
║uthentication.basic.username │ │ │ ║
║spring.cloud.dataflow.client.a│The login password. │<none> │java.lang.String ║
║uthentication.basic.password │ │ │ ║
║trigger.max-period │The maximum polling period in │30000 │java.lang.Integer ║
║ │milliseconds. Will be set to │ │ ║
║ │period if period > maxPeriod. │ │ ║
║trigger.period │The polling period in │1000 │java.lang.Integer ║
║ │milliseconds. │ │ ║
║trigger.initial-delay │The initial delay in │1000 │java.lang.Integer ║
║ │milliseconds. │ │ ║
║spring.cloud.dataflow.client.s│Skip Ssl validation. │true │java.lang.Boolean ║
║kip-ssl-validation │ │ │ ║
║spring.cloud.dataflow.client.e│Enable Data Flow DSL access. │false │java.lang.Boolean ║
║nable-dsl │ │ │ ║
║spring.cloud.dataflow.client.s│The Data Flow server URI. │http://localhost:9393 │java.lang.String ║
║erver-uri │ │ │ ║
╚══════════════════════════════╧══════════════════════════════╧══════════════════════════════╧══════════════════════════════╝
----
+
Since we configured the SCDF server to bind all stream and task apps to the `nfs` service, no deployment parameters are required.
+
[source,console,options=nowrap]
----
dataflow:>stream create inboundSftp --definition "sftp --username=<user> --password=<pass> --host=<host> --allow-unknown-keys=true --remote-dir=remote-files --local-dir=/var/scdf/shared-files/ --task.launch.request.taskName=fileIngestTask | task-launcher --spring.cloud.dataflow.client.server-uri=http://<dataflow-server-route>"
Created new stream 'inboundSftp'
dataflow:>stream deploy inboundSftp
Deployment request has been sent for stream 'inboundSftp'
----
+
Alternatively, we can bind the `nfs` service to the `fileIngestTask` by passing deployment properties to the task via the task launch request in the stream definition: `--task.launch.request.deployment-properties=deployer.*.cloudfoundry.services=nfs`
+
[source, console, options=nowrap]
----
dataflow:>stream deploy inboundSftp --properties "deployer.sftp.cloudfoundry.services=nfs"
----
. 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 --task.launch.request.deployment-properties='deployer.*.cloudfoundry.services=nfs' --password='******' --host=<host> │The stream has ║
║ │--remote-dir=remote-files --local-dir=/var/scdf/shared-files/ --task.launch.request.taskName=fileIngestTask --allow-unknown-keys=true │been successfully ║
║ │--username=<user> | task-launcher --spring.cloud.dataflow.client.server-uri=http://<dataflow-server-route> │deployed ║
╚═══════════╧═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╧══════════════════╝
----
+
. Inspect 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:
+
[source, console, options=nowrap]
----
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` application can be viewed by:
+
```
cf logs dataflow-server-N5RYLDj-inboundSftp-task-launcher --recent
```
. Add data
+
Sample data can be found in the `data/` directory of the <<Batch File Ingest>> project.
Connect to the SFTP server and upload `data/name-list.csv` into the `remote-files` directory.
Copy `data/name-list.csv` into the `/tmp/remote-files` directory which the SFTP source is monitoring.
When this file is detected, the `sftp` source will download it to the `/var/scdf/shared-files` directory specified by `--local-dir`, and emit a Task Launch Request.
The Task Launch Request includes the name of the task to launch along with the local file path, given as a command line argument.
Spring Batch binds each command line argument to a corresponding JobParameter.
The FileIngestTask job processes the file given by the JobParameter named `localFilePath`.
The `task-launcher` sink polls for messages using an exponential back-off.
Since there have not been any recent requests, the task will launch within 30 seconds after the request is published.
+
. 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 │Created ║
╚═══╧═══════╧═════════╧════════════════════════════╧═════════════════════╧══════════════════╝
----
+
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 │Wed Oct 31 03:17:34 EDT 2018 ║
║Start Time │Wed Oct 31 03:17:34 EDT 2018 ║
║End Time │Wed Oct 31 03:17:34 EDT 2018 ║
║Running │false ║
║Stopping │false ║
║Step Execution Count │1 ║
║Execution Status │COMPLETED ║
║Exit Status │COMPLETED ║
║Exit Message │ ║
║Definition Status │Created ║
║Job Parameters │ ║
║-spring.cloud.task.executionid(STRING) │1 ║
║run.id(LONG) │1 ║
║localFilePath(STRING) │/var/scdf/shared-files/name_list.csv║
╚═══════════════════════════════════════════╧════════════════════════════════════╝
----
+
. Verify data
+
When the the batch job runs, it processes the file in the local directory `/var/scdf/shared-files` and transforms each item to uppercase names and inserts it into the database.
+
Use https://github.com/pivotal-cf/PivotalMySQLWeb[PivotalMySQLWeb] to inspect the data.

BIN
src/main/asciidoc/images/metadata_store_1.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 36 KiB

BIN
src/main/asciidoc/images/people_table_1.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 71 KiB

BIN
src/main/asciidoc/images/people_table_2.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 64 KiB

2
src/main/asciidoc/index-docinfo.xml
View File

@@ -1,7 +1,7 @@
<productname>Spring Cloud Data Flow Samples</productname>
<releaseinfo>{project-version}</releaseinfo>
<copyright>
<year>2013-2017</year>
<year>2013-2018</year>
<holder>Pivotal Software, Inc.</holder>
</copyright>
<legalnotice>

2
src/main/asciidoc/index.adoc
View File

@@ -19,7 +19,7 @@ ifdef::backend-html5[]
Version {project-version}
(C) 2012-2017 Pivotal Software, Inc.
(C) 2012-2018 Pivotal Software, Inc.
_Copies of this document may be made for your own use and for distribution to
others, provided that you do not charge any fee for such copies and further

5
src/main/asciidoc/streaming/cassandra/http-to-cassandra/pcf.adoc
View File

@@ -21,7 +21,9 @@ CREATE TABLE book (
* The Spring Cloud Data Flow Cloud Foundry Server
include::{docs_dir}/cloudfoundry-server.adoc[]
===== Building and Running the Demo
===== Running the Demo
The source code for the <<Batch File Ingest>> batch job is located in `batch/file-ingest`
. https://github.com/spring-cloud/spring-cloud-dataflow/blob/master/spring-cloud-dataflow-docs/src/main/asciidoc/streams.adoc#register-a-stream-app[Register] the out-of-the-box applications for the Rabbit binder
+
@@ -32,7 +34,6 @@ include::{docs_dir}/maven-access.adoc[]
dataflow:>app import --uri {app-import-rabbit-maven}
```
+
+
. Create the stream
+

2
src/main/asciidoc/tasks/file-ingest/main.adoc
View File

@@ -66,7 +66,7 @@ dataflow:>
. Launch the task
+
```
dataflow:>task launch fileIngestTask --arguments "filePath=classpath:data.csv --spring.cloud.task.closecontext_enable=false"
dataflow:>task launch fileIngestTask --arguments "localFilePath=classpath:data.csv"
Launched task 'fileIngestTask'
dataflow:>
```