# FTP

Provide the remote directories and on specified intervals, the list of files in the directories is refreshed. Files are downloaded when they were not known before, or when their timestamp or size are changed. Only files with a timestamp younger than the specified maximum age are considered. Hashes of the files are maintained and used to check for content changes. Changed files are then fed into Kafka, either as a whole (update) or only the appended part (tail), depending on the configuration. Optionally, file bodies can be transformed through a pluggable system prior to putting them into Kafka.

## Connector Class

```
io.lenses.streamreactor.connect.ftp.source.FtpSourceConnector
```

## Example

{% hint style="success" %}
For more examples see the [tutorials](https://docs.lenses.io/latest/connectors/tutorials).
{% endhint %}

```bash
name=ftp-source
connector.class=io.lenses.streamreactor.connect.ftp.source.FtpSourceConnector
tasks.max=1

#server settings
connect.ftp.address=localhost:21
connect.ftp.user=ftp
connect.ftp.password=ftp

#refresh rate, every minute
connect.ftp.refresh=PT1M

#ignore files older than 14 days.
connect.ftp.file.maxage=P14D

#monitor /forecasts/weather/ and /logs/ for appends to files.
#any updates go to the topics `weather` and `error-logs` respectively.
connect.ftp.monitor.tail=/forecasts/weather/:weather,/logs/:error-logs

#keep an eye on /statuses/, files are retrieved as a whole and sent to topic `status`
connect.ftp.monitor.update=/statuses/:status

#keystyle controls the format of the key and can be string or struct.
#string only provides the file name
#struct provides a structure with the filename and offset
connect.ftp.keystyle=struct
```

## Data types <a href="#data-types" id="data-types"></a>

Each Kafka record represents a file and has the following types.

* The format of the keys is configurable through **connect.ftp.keystyle=string|struct**. It can be a string with the file name, or a FileInfo structure with the name: string and offset: long. The offset is always 0 for files that are updated as a whole, and hence only relevant for tailed files.
* The values of the records contain the body of the file as bytes.

## Tailing Versus Update as a Whole <a href="#tailing-versus-update-as-a-whole" id="tailing-versus-update-as-a-whole"></a>

The following rules are used.

Tailed files are only allowed to grow. Bytes that have been appended to it since the last inspection are yielded. Preceding bytes are not allowed to change;

Updated files can grow, shrink and change anywhere. The entire contents are yielded.

## Data converters <a href="#data-converters" id="data-converters"></a>

Instead of dumping whole file bodies (and the danger of exceeding Kafka’s message.max.bytes), one might want to give an interpretation to the data contained in the files before putting it into Kafka. For example, if the files that are fetched from the FTP are comma-separated values (CSVs), one might prefer to have a stream of CSV records instead. To allow to do so, the connector provides a pluggable conversion of SourceRecords. Right before sending a SourceRecord to the Connect framework, it is run through an object that implements:

```scala
package io.lenses.streamreactor.connect.ftp

trait SourceRecordConverter extends Configurable {
    def convert(in:SourceRecord) : java.util.List[SourceRecord]
}

```

The default object that is used is a pass-through converter, an instance of:

```scala
class NopSourceRecordConverter extends SourceRecordConverter{
    override def configure(props: util.Map[String, _]): Unit = {}
    override def convert(in: SourceRecord): util.List[SourceRecord] = Seq(in).asJava
}
```

To override it, create your own implementation of SourceRecordConverter and place the jar in the **`plugin.path`**.

```bash
connect.ftp.sourcerecordconverter=your.name.space.YourConverter
```

{% hint style="info" %}
To learn more examples of using the FTP Kafka connector read this [blog.](https://lenses.io/blog/2017/02/ftp-to-kafka/)
{% endhint %}

## Option Reference <a href="#options" id="options"></a>

<table data-full-width="true"><thead><tr><th width="244">Name</th><th width="235.5">Description</th><th width="110">Type</th><th>Default Value</th></tr></thead><tbody><tr><td>connect.ftp.address</td><td>host[:port] of the ftp server</td><td>string</td><td></td></tr><tr><td>connect.ftp.user</td><td>Username to connect with</td><td>string</td><td></td></tr><tr><td>connect.ftp.password</td><td>Password to connect with</td><td>string</td><td></td></tr><tr><td>connect.ftp.refresh</td><td>iso8601 duration that the server is polled</td><td>string</td><td></td></tr><tr><td>connect.ftp.file.maxage</td><td>iso8601 duration for how old files can be</td><td>string</td><td></td></tr><tr><td>connect.ftp.keystyle</td><td>SourceRecord keystyle, string or struct</td><td>string</td><td></td></tr><tr><td>connect.ftp.protocol</td><td>Protocol to use, FTP or FTPS</td><td>string</td><td>ftp</td></tr><tr><td>connect.ftp.timeout</td><td>Ftp connection timeout in milliseconds</td><td>int</td><td>30000</td></tr><tr><td>connect.ftp.filter</td><td>Regular expression to use when selecting files for processing</td><td>string</td><td>.*</td></tr><tr><td>connect.ftp.monitor.tail</td><td>Comma separated lists of path:destinationtopic; tail of file to tracked</td><td>string</td><td></td></tr><tr><td>connect.ftp.monitor.update</td><td>Comma separated lists of path:destinationtopic; whole file is tracked</td><td>string</td><td></td></tr><tr><td>connect.ftp.monitor.slicesize</td><td>File slice size in bytes</td><td>int</td><td>-1</td></tr><tr><td>connect.ftp.fileconverter</td><td>File converter class</td><td>string</td><td>io.lenses.streamreactor.connect.ftp.source.SimpleFileConverter</td></tr><tr><td>connect.ftp.sourcerecordconverter</td><td>Source record converter class</td><td>string</td><td>io.lenses.streamreactor.connect.ftp.source.NopSourceRecordConverter</td></tr><tr><td>connect.ftp.max.poll.records</td><td>Max number of records returned per poll</td><td>int</td><td>10000</td></tr></tbody></table>