# FTPS Source Connector for Confluent Platform The Kafka Connect FTPS connector is used to integrate with FTPS servers. The Kafka Connect FTPS Source connector provides the capability to watch a directory on an FTPS server for files and read the data as new files are written to the FTPS input directory. Each record in the input file is converted based on the user-supplied schema or an auto-generated schema and sent to a Apache Kafka® topic. ## Features The FTPS Source connector includes the following features: - [At least once delivery](#ftps-source-at-least-once-delivery) - [Supported file types](#ftps-source-supported-file-types) - [Multiple tasks](#ftps-source-multiple-tasks) - [File pattern-based filtering](#ftps-source-pattern-based-filtering) - [Automatic schema generation](#ftps-source-automatic-schema-generation) ### At least once delivery This connector guarantees that records are delivered to the Kafka topic at least once. If the connector restarts, there may be some duplicate records in the Kafka topic. ### Supported file types The connector currently supports parsing CSV, JSON, and TXT files using the same connector. It uses a file’s extension to figure out the file type and the parser to be used. The file types to be parsed can be specified using a config parameter. ### Multiple tasks The connector supports running multiple tasks simultaneously, which can be configured with the `tasks.max` configuration property. This can lead to huge performance gains when multiple files need to be parsed. ### File pattern-based filtering The connector supports filtering files based on a regular expression. This can also be used to parse only specific kinds of files. For example, setting `ftps.input.file.pattern` to `test*+\.(csv|json)` causes the connector to parse only CSV and JSON files beginning with `test`, while ignoring all other files. ### Automatic schema generation This is supported for CSV and JSON files. The first record in a file is used to generate the schema for the file. ## Limitations * Automatic schema generation is not supported for nested JSON. Instead, schemaless JSON parsing can be used by setting `ftps.schemaless.json.parsing` to `true`. ## Install the FTPS Source Connector The FTPS connector is compatible with Confluent Platform from version 4.1 (and later). Prior versions do not work with this connector. You can install this connector by using the [confluent connect plugin install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command, or by manually downloading the ZIP file. ### Prerequisites - You must install the connector on every machine where Connect will run. - If you want to install the connector using Confluent Marketplace, you must install the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This is installed by default with Confluent Enterprise. - Kafka Broker: Confluent Platform 3.3.0 or later. - Connect: Confluent Platform 4.1.0 or later. - Java 1.8. - All the required directories and files on the FTPS server must be accessible by the connector. - The schema can change with each file. For this reason, Confluent recommends setting the topic schema property to `NONE`. ### Install the connector using the Confluent CLI To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory and run the following command: ```bash confluent connect plugin install confluentinc/kafka-connect-ftps:latest ``` You can install a specific version by replacing `latest` with a version number as shown in the following example: ```bash confluent connect plugin install confluentinc/kafka-connect-ftps:1.0.4 ``` ## Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-ftps) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/install.html). ## License You can use this connector for a 30-day trial period without a license key. After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information. See [Confluent Licensing](connector_config.md#ftps-source-connector-license-config) for license properties and [License topic configuration](connector_config.md#ftps-source-license-topic-configuration) for information about the license topic. ## Configuration Properties For a complete list of configuration properties for this connector, see [Configuration Reference for FTPS Source Connector for Confluent Platform](connector_config.md#ftps-source-connector-config). For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see [Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster). ## Quick Start In this Quick Start, you will configure the FTPS Source connector to copy data from files on an FTPS server to the Kafka topic. 1. Install the connector through the [Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html). ```bash # run from your confluent platform installation directory confluent connect plugin install confluentinc/kafka-connect-ftps:latest ``` 2. Start the Confluent Platform. ```bash confluent local start ``` 3. Check the status of all services. ```bash confluent local services status ``` 4. Configure your connector by first creating a `.properties` file named `ftps.properties` with the following properties. ```bash // substitute <> with your config name=FTPSConnector connector.class=io.confluent.connect.ftps.FtpsSourceConnector key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter tasks.max=3 kafka.topic= confluent.topic.bootstrap.servers=localhost:9092 confluent.topic.replication.factor=1 confluent.license= batch.size=100 ftps.input.path= ftps.host= ftps.username= ftps.password= ftps.ssl.key.password= ftps.ssl.keystore.location= ftps.ssl.keystore.password= ftps.ssl.truststore.location= ftps.ssl.truststore.password= ``` 5. Start the FTPS Source connector by loading the connector’s configuration with the following command: ```bash confluent local load ftps --config ftps.properties ``` 6. Confirm that the connector is in a `RUNNING` state. ```bash confluent local status FTPSConnector ``` 7. Confirm the messages were delivered to the configured topic in Kafka. ```bash confluent local consume --from-beginning ``` ## Examples ### JSON file with schema This example follows the same steps as the Quick Start. Review the above Quick Start for help running the Confluent Platform and installing the FTPS connector package. 1. Generate a JSON file using the command below and push it to FTPS server: ```bash echo $'{"id":1,"first_name":"Roscoe","last_name":"Brentnall","email":"rbrentnall0@mediafire.com","gender":"Male","ip_address":"202.84.142.254","last_login":"2018-02-12T06:26:23Z","account_balance":1450.68,"country":"CZ","favorite_color":"#4eaefa"}\n{"id":2,"first_name":"Gregoire","last_name":"Fentem","email":"gfentem1@nsw.gov.au","gender":"Male","ip_address":"221.159.106.63","last_login":"2015-03-27T00:29:56Z","account_balance":1392.37,"country":"ID","favorite_color":"#e8f686"}' > "json-ftps-source.json" ``` 2. Create an `ftps.properties` file with the following contents: ```properties name=JsonFTPS tasks.max=1 connector.class=io.confluent.connect.ftps.FtpsSourceConnector key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter kafka.topic=ftps-json-topic ftps.input.file.pattern=json-ftps-source\.json ftps.input.path= ftps.host= ftps.username= ftps.password= ftps.port= schema.generation.enabled=false ftps.ssl.key.password= ftps.ssl.keystore.location= ftps.ssl.keystore.password= ftps.ssl.truststore.location= ftps.ssl.truststore.password= 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 FTPS Source connector. ```bash confluent local load JsonFTPS --config ftps.properties ``` #### IMPORTANT Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments. ### CSV file with automatic schema generation This example follows the same steps as the Quick Start. Review the above Quick Start for help running the Confluent Platform and installing the FTPS connector package. 1. Generate a CSV file using the command below and push it to FTPS server: ```bash echo $'119736,FL,CLAY COUNTY,498960,498960,498960,498960,498960,792148.9,0,9979.2,0,0,30.102261,-81.711777,Residential,Masonry,1\n291781,FL,NASSAU COUNTY,0,723734.17,0,0,723734.17,955908.09,0,0,0,0,30.53674,-81.77496,Residential,Masonry,3' > "csv-ftps-source.csv" ``` 2. Create an `ftps.properties` file with the following contents: ```properties name=CsvFTPS tasks.max=1 connector.class=io.confluent.connect.ftps.FtpsSourceConnector key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter kafka.topic=ftps-csv-topic ftps.input.file.pattern=csv-ftps-source\.csv ftps.input.path= ftps.host= ftps.port= ftps.username= ftps.password= schema.generation.enabled=true ftps.ssl.key.password= ftps.ssl.keystore.location= ftps.ssl.keystore.password= ftps.ssl.truststore.location= ftps.ssl.truststore.password= ``` 3. Load the FTPS Source connector. ```bash confluent local load CsvFTPS --config ftps.properties ``` #### IMPORTANT Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments. ### TXT file This example follows the same steps as the Quick Start. Review the above Quick Start for help running the Confluent Platform and installing the FTPS connector package. 1. Generate a TXT file using the command below and push it to FTPS server: ```bash echo $'This is line 1.\nThis is line 2.\nThis is line 3.' > "txt-ftps-source.txt" ``` 2. Create an `ftps.properties` file with the following contents: ```properties name=TxtFTPS tasks.max=1 connector.class=io.confluent.connect.ftps.FtpsSourceConnector key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter kafka.topic=ftps-txt-topic ftps.input.file.pattern=txt-ftps-source\.txt ftps.input.path= ftps.host= ftps.port= ftps.username= ftps.password= ftps.ssl.key.password= ftps.ssl.keystore.location= ftps.ssl.keystore.password= ftps.ssl.truststore.location= ftps.ssl.truststore.password= ``` 3. Load the FTPS Source connector. ```bash confluent local load TxtFTPS --config ftps.properties ``` #### IMPORTANT Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments. ### Parsing multiple file types at once This example follows the same steps as the Quick Start. Review the above Quick Start for help running the Confluent Platform and installing the FTPS connector package. 1. Generate a JSON file using the command below and push it to FTPS server: ```bash echo $'{"id":1,"first_name":"Roscoe","last_name":"Brentnall","email":"rbrentnall0@mediafire.com","gender":"Male","ip_address":"202.84.142.254","last_login":"2018-02-12T06:26:23Z","account_balance":1450.68,"country":"CZ","favorite_color":"#4eaefa"}\n{"id":2,"first_name":"Gregoire","last_name":"Fentem","email":"gfentem1@nsw.gov.au","gender":"Male","ip_address":"221.159.106.63","last_login":"2015-03-27T00:29:56Z","account_balance":1392.37,"country":"ID","favorite_color":"#e8f686"}' > "json-ftps-source.json" ``` 2. Generate a CSV file using the command below and push it to FTPS server: ```bash echo $'119736,FL,CLAY COUNTY,498960,498960,498960,498960,498960,792148.9,0,9979.2,0,0,30.102261,-81.711777,Residential,Masonry,1\n291781,FL,NASSAU COUNTY,0,723734.17,0,0,723734.17,955908.09,0,0,0,0,30.53674,-81.77496,Residential,Masonry,3' > "csv-ftps-source.csv" ``` 3. Generate a TXT file using the command below and push it to FTPS server: ```bash echo $'This is line 1.\nThis is line 2.\nThis is line 3.' > "txt-ftps-source.txt" ``` 4. The goal is to create a connector which parses only files having a filename with the following format: ```text -ftps-source. ``` where `` is the type of file (for example, CSV, TXT, or JSON). Schema will be automatically generated for CSV and JSON files. 5. Create an `ftps.properties` file with the following contents: ```properties name=MixedFTPS tasks.max=3 kafka.topic=ftps-mixed-topic connector.class=io.confluent.connect.ftps.FtpsSourceConnector key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter ftps.input.file.pattern=.+-ftps-source\..+ ftps.input.path= ftps.host= ftps.username= ftps.password= ftps.port= ftps.ssl.key.password= ftps.ssl.keystore.location= ftps.ssl.keystore.password= ftps.ssl.truststore.location= ftps.ssl.truststore.password= ``` 6. Load the FTPS Source connector. ```bash confluent local load MixedFTPS --config ftps.properties ``` #### IMPORTANT Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.