<a id="sftp-csv-source-connector"></a>

# CSV Source Connector for Confluent Platform

The Kafka Connect CSV Source connector monitors the SFTP directory specified in `input.path` for files and reads them as CSVs, converting each of the records to the strongly typed equivalent specified in `key.schema` and `value.schema`. The connector can also auto generate the `key.schema` and `value.schema` at run time if `schema.generation.enabled` is true.

To use this connector, specify the name of the connector class in the `connector.class` configuration property.

```properties
connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
```

<a id="sftp-csv-source-connector-limitations"></a>

## Limitations

The CSV Source connector will only authenticate with Kerberos when a keytab is
cached on the system–not when the keytab is provided in the configs.
Additionally, the Connector will only work with Kerberos on Confluent Platform 5.5.4
or later.

<a id="sftp-csv-source-connector-example"></a>

## CSV Source Connector Examples

The following examples follow the same steps as the Quick Start for installing Confluent Platform and the SFTP connector package. Review the [Quick Start](overview.md#sftp-source-connector-quickstart) for help running the Confluent Platform and installing the SFTP connector package.

<a id="sftp-connector-csv-with-schema-example"></a>

### CSV with Schema Example

This example reads CSV files and writes them to Kafka. It parses them using the schema specified in `key.schema` and `value.schema`.

1. Set up an SFTP data directory for files to process, generate test data in your local and push it to the SFTP server. For example:
   ```none
   echo $'1,Salmon,Baitman,sbaitman0@feedburner.com,Male,120.181.75.98,2015-03-01T06:01:15Z,17462.66,IT,#f09bc0\n2,Debby,Brea,dbrea1@icio.us,Female,153.239.187.49,2018-10-21T12:27:12Z,14693.49,CZ,#73893a' > "csv-sftp-source.csv"
   ```
2. Create an `sftp.properties` file with the following contents:
   ```properties
   name=CsvSchemaSftp
   kafka.topic=sftp-testing-topic
   tasks.max=1
   connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
   cleanup.policy=NONE
   behavior.on.error=IGNORE
   input.path=/path/to/data
   error.path=/path/to/error
   finished.path=/path/to/finished
   input.file.pattern=csv-sftp-source.csv
   sftp.username=username
   sftp.password=password
   sftp.host=localhost
   sftp.port=22
   csv.first.row.as.header=false
   key.schema={\"name\" : \"com.example.users.UserKey\",\"type\" : \"STRUCT\",\"isOptional\" : false,\"fieldSchemas\" : {\"id\" : {\"type\" : \"INT64\",\"isOptional\" : false}}}
   value.schema={\"name\" : \"com.example.users.User\",\"type\" : \"STRUCT\",\"isOptional\" : false,\"fieldSchemas\" : {\"id\" : {\"type\" : \"INT64\",\"isOptional\" : false},\"first_name\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"last_name\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"email\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"gender\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"ip_address\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"last_login\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"account_balance\" : {\"name\" : \"org.apache.kafka.connect.data.Decimal\",\"type\" : \"BYTES\",\"version\" : 1,\"parameters\" : {\"scale\" : \"2\"},\"isOptional\" : true},\"country\" : {\"type\" : \"STRING\",\"isOptional\" : true},\"favorite_color\" : {\"type\" : \"STRING\",\"isOptional\" : true}}}
   ```
3. Load the SFTP CSV Source connector.
   ```bash
   confluent local load CsvSchemaSftp --config sftp.properties
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.
4. Validate messages are sent to Kafka serialized with Avro.
   ```bash
   kafka-avro-console-consumer --topic sftp-testing-topic --from-beginning --bootstrap-server localhost:9092
   ```

<a id="sftp-connector-tsv-example"></a>

### TSV Input File Example

The following example loads a TSV file and produces each record to Kafka. TSV Source connector works just like the CSV Source connector, except that you have to provide the extra configuration property `csv.separator.char=9`. The number 9 is the ASCII value for tab space.

1. Generate a TSV data set using the command below:
   ```bash
   echo $'id\tfirst_name\tlast_name\temail\tgender\tip_address\tlast_login\taccount_balance\tcountry\tfavorite_color\n1\tPadraig\tOxshott\tpoxshott0@dion.ne.jp\tMale\t47.243.121.95\t2016-06-24T22:43:42Z\t15274.22\tJP\t#06708f\n2\tEdi\tOrrah\teorrah1@cafepress.com\tFemale\t158.229.234.101\t2017-03-01T17:52:47Z\t12947.6\tCN\t#5f2aa2' > "tsv-sftp-source.tsv"
   ```
2. Create an `sftp.properties` file with the following contents:
   ```properties
   name=TsvSftp
   tasks.max=1
   connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
   input.path=/path/to/data
   error.path=/path/to/error
   finished.path=/path/to/finished
   cleanup.policy=NONE
   input.file.pattern=tsv-sftp-source.tsv
   behavior.on.error=IGNORE
   sftp.username=username
   sftp.password=password
   sftp.host=localhost
   sftp.port=22
   kafka.topic=sftp-tsv-topic
   schema.generation.enabled=true
   csv.first.row.as.header=true
   csv.separator.char=9
   ```
3. Load the SFTP TSV Source connector.
   ```bash
   confluent local load TsvSftp --config sftp.properties
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.

### Key-Based TLS Authentication Example

The following connector configuration example shows the configuration properties
used when the connector requires TLS private/public key authentication:

```properties
name=sftp-source-connector
topics=sftptopic
tasks.max=1
connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
tls.private.key = <tls-private-key>
tls.public.key = <tls-public-key>
#tls.passphrase=<uncomment-if-required>
sftp.username=<sftp-username>
confluent.topic.bootstrap.servers=localhost:9092
sftp.host=192.168.2.5
sftp.port=22
flush.size=3
format.class=io.confluent.connect.avro.AvroConverter
sftp.working.dir= /home/<path-to-files/
```

Before configuring the `tls.private.key` property, see the example
shown in the [Security](license_config.md#sftp-source-security-configs) section of the
[SFTP Source Connector Configuration Properties](license_config.md#sftp-source-connector-license-config).

<a id="sftp-csv-source-connector-config"></a>

## Configuration Properties

For connector-specific configuration properties, see
[SFTP Source Connector Configuration Properties](license_config.md#sftp-source-connector-license-config)