[[spring-cloud-data-flow-samples-sftp-file-ingest-overview]]
:sectnums:
:docs_dir: ../..
=== Batch File Ingest - SFTP Demo

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.

The source for the demo project is located in the `batch/file-ingest` directory at the top-level of this repository.

==== Prerequisites
* A Running Data Flow Shell
include::{docs_dir}/shell.adoc[]

include::local.adoc[]

include::pcf.adoc[]

include::kubernetes.adoc[]

==== Limiting Concurrent Task Executions

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.

Fortunately, Spring Cloud Data Flow provides configuration settings to https://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.

===== Configuring the SCDF server
Set the maximum concurrent tasks to 3.
For running tasks on a local server, restart the server, adding a command line argument `spring.cloud.dataflow.task.platform.local.accounts[default].maximum-concurrent-tasks=3`.

If running on Cloud Foundry, `cf set-env <dataflow-server> SPRING_CLOUD_DATAFLOW_TASK_PLATFORM_CLOUDFOUNDRY_ACCOUNTS[DEFAULT]_DEPLOYMENT_MAXIMUMCONCURRENTTASKS 3`, and restage.

===== Running the demo
Follow the main demo instructions but change the `Add Data` step, as described below.

. Monitor the task launcher
+
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.,

```
sftp>cd remote-files
sftp>lcd batch/file-ingest/data/split
sftp>mput *
```
Or if using the local machine as the SFTP server:
```
>cp * /tmp/remote-files
```
In the `task-launcher` logs, you should now see:
[source, console, options=nowrap]
----
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.
...
----

==== 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
+
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:
+
```
$ ./mvnw clean install -DskipTests -PgenerateApps
$ cd apps/sftp-dataflow-source-<binder>
```
+
Add the following dependencies to `pom.xml`:
+
```
<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>
```
+
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:
+
```
spring.datasource.url=jdbc:h2:tcp://localhost:19092/mem:dataflow
```
+
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`:
+
```
spring.integration.jdbc.initialize-schema=always
```
+
Build the app:
+
```
$./mvnw clean package
```
. Register the jar
+
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:
+
```
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
+
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:
+
* 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
+
If you are running in Cloud Foundry, set the deployment properties to bind `sftp` to the `mysql` service. For example:
+
```
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:
+
```
sftp>cd remote-files
sftp>lcd batch/file-ingest/data/split
sftp>put names_aa.csv
```
+
Or if using the local machine as the SFTP server:
+
```
$cp names_aa.csv truncate INT_METADATA_STORE;
```
. Inspect data
+
Using a Database browser, as described in the main demo, view the contents of the `INT_METADATA_STORE` table.
+
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:
+
```
$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:
+
image::people_table_2.png[title="People Data with Duplicates"]
+
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 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
