# SFTP Sink Connector for Confluent Platform The SFTP Sink connector for Kafka Connect exports records from Apache Kafka® topics to files on an SSH File Transfer Protocol (SFTP) server, in CSV/TSV, Avro, JSON, or Parquet format. The connector polls Kafka for new records and writes them to SFTP files, using a partitioner to split each Kafka partition’s data into chunks. Each chunk becomes a separate file whose name encodes the topic, the Kafka partition, and the start offset. By default, the connector uses the partitioner that preserves Kafka partitioning. The size of each chunk depends on the number of records written and on schema compatibility. ## SFTP Sink connector features The SFTP Sink Connector includes the following features: - [Exactly once delivery](#sftp-sink-exactly-once-delivery) - [Dead Letter Queue](#sftp-sink-dead-letter-queue) - [Multiple tasks](#sftp-sink-multiple-tasks) - [Pluggable data format with or without schema](#sftp-sink-pluggable-data-format) - [Pluggable partitioner](#sftp-sink-pluggable-partitioner) - [Client-side encryption](#sftp-sink-csfle-sm) ### Exactly once delivery Records that are exported using a deterministic partitioner are delivered with exactly-once semantics regardless of the eventual consistency of SFTP. To learn more, see [Exactly-once delivery on top of eventual consistency](#s3-exactly-once). ### Dead Letter Queue This connector supports the Dead Letter Queue (DLQ) functionality. For information about accessing and using the DLQ, see [Confluent Platform Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue). ### Multiple tasks The SFTP Sink Connector supports running one or more tasks. You can specify the number of tasks in the `tasks.max` configuration parameter. This can lead to performance gains when multiple files need to be parsed. ### Pluggable data format with or without schema Out of the box, the connector supports writing data to SFTP files in Avro, CSV/TSV, JSON and parquet formats. Besides records with schema, the connector supports exporting plain JSON records without schema in text files, one record per-line. In general, the connector may accept any format that provides an implementation of the `Format` interface. ### Pluggable partitioner The connector comes out of the box with partitioners that support default partitioning based on Kafka partitions, field partitioning, and time-based partitioning in days or hours. You may implement your own partitioners by extending the `Partitioner` class. Additionally, you can customize time based partitioning by extending the `TimeBasedPartitioner` class. * **Schema Evolution:** Schema evolution only works if the records are generated with the default naming strategy, which is `TopicNameStrategy`. An error may occur if other naming strategies are used. This is because records are not compatible with each other. `schema.compatibility` should be set to `NONE` if other naming strategies are used. This may result in small object files because the sink connector creates a new file every time the schema ID changes between records. See [Subject Name Strategy](/platform/current/schema-registry/serdes-develop/index.html) for more information about naming strategies. ### Client-side encryption This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html). ## Limitations The SFTP Sink connector does not support the [RegexRouter](https://docs.confluent.io/platform/current/connect/transforms/regexrouter.html#regexrouter) SMT. ## 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 Platform license](configuration_options.md#sftp-sink-connector-license-config) for license properties and [License topic configuration](configuration_options.md#sftp-sink-license-topic-configuration) for information about the license topic. ## SFTP Sink connector configuration properties For a complete list of configuration properties for the sink connector, see [Configuration Reference for SFTP Sink Connector for Confluent Platform](configuration_options.md#sftp-sink-configuration-options). 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). ## Install the SFTP Sink 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, or Kafka 0.11.0 or later. - Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later. - Java 1.8. ### 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-sftp: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-sftp:3.2.0 ``` ### Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-sftp) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/install.html). ## Quick Start This quick start uses the SFTP Sink Connector to export data produced by the Avro console producer to SFTP directory. First, start all the necessary services using Confluent CLI. ```bash confluent local start ``` Every service will start in order, printing a message with its status: ```bash Starting Zookeeper Zookeeper is [UP] Starting Kafka Kafka is [UP] Starting Schema Registry Schema Registry is [UP] Starting Kafka REST Kafka REST is [UP] Starting Connect Connect is [UP] Starting KSQL Server KSQL Server is [UP] Starting Control Center Control Center is [UP] ``` Next, start the Avro console producer to import a few records to Kafka: ```bash ./bin/kafka-avro-console-producer --broker-list localhost:9092 --topic test_sftp_sink \ --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' ``` In the console producer, enter the following: ```bash {"f1": "value1"} {"f1": "value2"} {"f1": "value3"} ``` The three records entered are published to the Kafka topic `test_sftp_sink` in Avro format. Before starting the connector, ensure the configurations in `etc/kafka-connect-sftp/quickstart-sftp.properties` are properly set to your configurations of SFTP (for example, `sftp.hostname` must point to the proper SFTP host). Then, start connector by loading its configuration with the following command: ```bash confluent local load sftp-sink --config etc/kafka-connect-sftp/quickstart-sftp.properties { "name": "sftp-sink", "config": { "topics": "test_sftp_sink", "tasks.max": "1", "connector.class": "io.confluent.connect.sftp.SftpSinkConnector", "confluent.topic.bootstrap.servers": "localhost:9092", "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner", "schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator", "flush.size": "3", "schema.compatibility": "NONE", "format.class": "io.confluent.connect.sftp.sink.format.avro.AvroFormat", "storage.class": "io.confluent.connect.sftp.sink.storage.SftpSinkStorage", "sftp.host": "localhost", "sftp.port": "2222", "sftp.username": "foo", "sftp.password": "pass", "sftp.working.dir": "/share", "name": "sftpconnector" }, "tasks": [] } ``` To check that the connector started successfully, view the Connect worker’s log by entering: ```bash confluent local services connect log ``` Towards the end of the log you should see that the connector starts, logs a few messages, and then exports data from Kafka to SFTP. Once the connector finishes ingesting data to SFTP, check that the data is available in the SFTP working directory: You should see a file with name `/topics/test_sftp_sink/partition=0/test_sftp_sink+0+0000000000.avro` The file name is encoded as `topic+kafkaPartition+startOffset+endOffset.format`. To extract the contents of the file, use `avro-tools-1.8.2.jar` (available in the [Apache Archives](http://archive.apache.org/dist/avro/avro-1.8.2/java/avro-tools-1.8.2.jar)). Move `avro-tools-1.8.2.jar` to SFTP’s working directory and run the following command: ```bash java -jar avro-tools-1.8.2.jar tojson //topics/test_sftp_sink/partition=0/test_sftp_sink+0+0000000000.avro ``` You should see the following output: ```bash {"f1":"value1"} {"f1":"value2"} {"f1":"value3"} ``` Finally, stop the Connect worker as well as all the rest of Confluent Platform by running: ```bash confluent local stop ``` Your output should resemble: ```none Stopping Control Center Control Center is [DOWN] Stopping KSQL Server KSQL Server is [DOWN] Stopping Connect Connect is [DOWN] Stopping Kafka REST Kafka REST is [DOWN] Stopping Schema Registry Schema Registry is [DOWN] Stopping Kafka Kafka is [DOWN] Stopping Zookeeper Zookeeper is [DOWN] ``` *Or*, stop all the services and additionally wipe out any data generated during this quick start by running: ```bash confluent local destroy ``` Your output should resemble: ```bash Stopping Control Center Control Center is [DOWN] Stopping KSQL Server KSQL Server is [DOWN] Stopping Connect Connect is [DOWN] Stopping Kafka REST Kafka REST is [DOWN] Stopping Schema Registry Schema Registry is [DOWN] Stopping Kafka Kafka is [DOWN] Stopping Zookeeper Zookeeper is [DOWN] Deleting: /var/folders/ty/rqbqmjv54rg_v10ykmrgd1_80000gp/T/confluent.PkQpsKfE ``` ### SFTP file names The SFTP data model is a flat structure. Each record gets stored into a file and the name of each file serves as the unique key. Generally, the hierarchy of files in which records get stored follow this format: ```bash ///++. ``` where: * `` is specified with the connector’s `sftp.working.dir` configuration property, which defaults to the literal value `topics` and helps create file with a name that don’t clash with existing file’s name in the same directory. * `` corresponds to the name of the Kafka topic from which the records were read. * `` is generated by the SFTP sink connector’s partitioner (see [Partitioning records into SFTP files](#sftp-partitioners)). * `` is the Kafka partition number from which the records were read. * `` is the Kafka offset of the first record written to this file. * `` is the extension identifing the format in which the records are serialized in this SFTP file. ### Partitioning records into SFTP files The SFTP sink connector’s *partitioner* determines how records read from a Kafka topic are partitioned into SFTP directories. The partitioner determines the `` portion of the SFTP file names (see [SFTP file names](#sftp-file-names)). The partitioner is specified in the connector configuration with the `partitioner.class` configuration property. The SFTP sink connector comes with the following partitioners: * **Default (Kafka) Partitioner**: The `io.confluent.connect.storage.partitioner.DefaultPartitioner` preserves the same topic partitions as in Kafka, and records from each topic partition ultimately end up in SFTP file with names that include the Kafka topic and Kafka partitions. The `` is always `/partition=`, resulting in SFTP file names of the form `//partition=/++.`. * **Field Partitioner**: The `io.confluent.connect.storage.partitioner.FieldPartitioner` determines the partition from the field within each record identified by the connector’s `partition.field.name` configuration property, which has no default. This partitioner requires `STRUCT` record type values. The `` is always `/=`, resulting in SFTP file names of the form `//=/++.`. * **Time-Based Partitioner**: The `io.confluent.connect.storage.partitioner.TimeBasedPartitioner` determines the partition from the year, month, day, hour, minutes, and/or seconds. This partitioner requires the following connector configuration properties: * The `path.format` configuration property specifies the pattern used for the `` portion of the SFTP file name. For example, when `path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH`, SFTP file names will have the form `//year=YYYY/month=MM/day=dd/hour=HH/++.`. * The `partition.duration.ms` configuration property defines the maximum granularity of the SFTP files within a single encoded partition directory. For example, setting `partition.duration.ms=600000` (10 minutes) will result in each SFTP file in that directory having no more than 10 minutes of records. * The `locale` configuration property specifies the JDK’s locale used for formatting dates and times. For example, use `en-US` for US English, `en-GB` for UK English, `fr-FR` for French (in France). These may vary by Java version; see the [available locales](http://www.localeplanet.com/java/). * The `timezone` configuration property specifies the current timezone in which the dates and times will be treated. Use standard short names for timezones such as `UTC` or (without daylight savings) `PST`, `EST`, and `ECT`, or longer standard names such as `America/Los_Angeles`, `America/New_York`, and `Europe/Paris`. These may vary by Java version; see the [available timezones within each locale](http://www.localeplanet.com/java), such as [those within the “en_US” locale](http://www.localeplanet.com/java/en-US/index.html). * The `timestamp.extractor` configuration property determines how to obtain a timestamp from each record. Values can include `Wallclock` (the default) to use the system time when the record is processed, `Record` to use the timestamp of the Kafka record denoting when it was produced or stored by the broker, `RecordField` to extract the timestamp from one of the fields in the record’s value as specified by the `timestamp.field` configuration property. * **Daily Partitioner**: The `io.confluent.connect.storage.partitioner.DailyPartitioner` is equivalent to the TimeBasedPartitioner with `path.format='year'=YYYY/'month'=MM/'day'=dd` and `partition.duration.ms=86400000` (one day, for one SFTP file in each daily directory). This partitioner always results in SFTP file names of the form `//year=YYYY/month=MM/day=dd/++.`. This partitioner requires the following connector configuration properties: * The `locale` configuration property specifies the JDK’s locale used for formatting dates and times. For example, use `en-US` for US English, `en-GB` for UK English, `fr-FR` for French (in France). These may vary by Java version; see the [available locales](http://www.localeplanet.com/java/). * The `timezone` configuration property specifies the current timezone in which the dates and times will be treated. Use standard short names for timezones such as `UTC` or (without daylight savings) `PST`, `EST`, and `ECT`, or longer standard names such as `America/Los_Angeles`, `America/New_York`, and `Europe/Paris`. These may vary by Java version; see the [available timezones within each locale](http://www.localeplanet.com/java), such as [those within the “en_US” locale](http://www.localeplanet.com/java/en-US/index.html). * The `timestamp.extractor` configuration property determines how to obtain a timestamp from each record. Values can include `Wallclock` (the default) to use the system time when the record is processed, `Record` to use the timestamp of the Kafka record denoting when it was produced or stored by the broker, `RecordField` to extract the timestamp from one of the fields in the record’s value as specified by the `timestamp.field` configuration property. * **Hourly Partitioner**: The `io.confluent.connect.storage.partitioner.HourlyPartitioner` is equivalent to the TimeBasedPartitioner with `path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH` and `partition.duration.ms=3600000` (one hour, for one SFTP file in each hourly directory). This partitioner always results in SFTP file names of the form `//year=YYYY/month=MM/day=dd/hour=HH/++.`. This partitioner requires the following connector configuration properties: * The `locale` configuration property specifies the JDK’s locale used for formatting dates and times. For example, use `en-US` for US English, `en-GB` for UK English, `fr-FR` for French (in France). These may vary by Java version; see the [available locales](http://www.localeplanet.com/java/). * The `timezone` configuration property specifies the current timezone in which the dates and times will be treated. Use standard short names for timezones such as `UTC` or (without daylight savings) `PST`, `EST`, and `ECT`, or longer standard names such as `America/Los_Angeles`, `America/New_York`, and `Europe/Paris`. These may vary by Java version; see the [available timezones within each locale](http://www.localeplanet.com/java), such as [those within the “en_US” locale](http://www.localeplanet.com/java/en-US/index.html). * The `timestamp.extractor` configuration property determines how to obtain a timestamp from each record. Values can include `Wallclock` (the default) to use the system time when the record is processed, `Record` to use the timestamp of the Kafka record denoting when it was produced or stored by the broker, `RecordField` to extract the timestamp from one of the fields in the record’s value as specified by the `timestamp.field` configuration property. As noted below, the choice of `timestamp.extractor` affects whether the SFTP sink connector can support exactly once delivery. You can also choose to use a custom partitioner by implementing the `io.confluent.connect.storage.partitioner.Partitioner` interface, packaging your implementation into a JAR file, and then: 1. Place the JAR file into the `share/java/kafka-connect-sftp` directory of your Confluent Platform installation **on each worker node**. 2. Restart all of the Connect worker nodes. #. Configure SFTP sink connectors to use your fully-qualified partitioner class name. ### SFTP File Formats The SFTP sink connector can serialize multiple records into each SFTP file using a number of formats. The connector’s `format.class` configuration property identifies the name of the Java class that implements the `io.confluent.connect.storage.format.Format` interface. The SFTP sink connector comes with several implementations: * **Avro**: Use `format.class=io.confluent.connect.sftp.sink.format.avro.AvroFormat` to write the SFTP file as an Avro container file and include the Avro schema in the container file followed by one or more records. The connector’s `avro.codec` configuration property specifies the Avro compression code, and values can be `null` (the default) for no Avro compression, `deflate` to use the deflate algorithm as specified in [RFC 1951](https://www.rfc-editor.org/info/rfc1951), `snappy` to use Google’s [Snappy](http://google.github.io/snappy/) compression library, and `bzip2` for BZip2 compression. Optionally set `enhanced.avro.schema.support=true` to enable enum symbol preservation and package name awareness. * **JSON**: Use `format.class=io.confluent.connect.sftp.sink.format.json.JsonFormat` to write the SFTP file containing one JSON serialized record per line. The connector’s `sftp.compression.type` configuration property can be set to `none` (the default) for no compression or `gzip` for GZip compression. * **Parquet**: Use `format.class=io.confluent.connect.sftp.sink.format.parquet.ParquetFormat` to write the SFTP file as a parquet container file and will include the Avro schema in the container file followed by one or more records. * **CSV**: Use `format.class=io.confluent.connect.sftp.sink.format.csv.CsvFormat` to write the SFTP file containing a comma-separated line for each record. * **TSV**: Use `format.class=io.confluent.connect.sftp.sink.format.tsv.TsvFormat` to write the SFTP file containing a tab space separated line for each record. You can also choose to use a custom partitioner by implementing the `io.confluent.connect.storage.format.Format` interface, packaging your implementation into a JAR file, and then: 1. Place the JAR file into the `share/java/kafka-connect-sftp` directory of your Confluent Platform installation **on each worker node**. 2. Restart all of the Connect worker nodes. #. Configure SFTP sink connectors with `format.class` set to the fully-qualified class name of your format implementation. ## Exactly-once delivery on top of eventual consistency The SFTP Sink Connector is able to provide exactly-once semantics to its consumers, under the condition that the connector is supplied with a deterministic partitioner. Currently, out of the available partitioners, the default and field partitioners are always deterministic. `TimeBasedPartitioner` can be deterministic with some configurations, discussed below. This implies that, when any of these partitioners is used, splitting of files always happens at the same offsets for a given set of Kafka records. These partitioners take into account `flush.size` and `schema.compatibility` to decide when to roll and save a new file to SFTP directories. The connector always delivers files in SFTP that contain the same records, even under the presence of failures. If a connector task fails before an upload completes, the file does not become visible to SFTP. If, on the other hand, a failure occurs after the upload has completed but before the corresponding offset is committed to Kafka by the connector, then a re-upload will take place. However, such a re-upload is transparent to the user of the SFTP, who at any time will have access to the same records made eventually available by successful uploads to SFTP files and directories. To guarantee exactly-once semantics with the `TimeBasedPartitioner`, the connector must be configured to use a deterministic implementation of `TimestampExtractor` and a deterministic rotation strategy. The deterministic timestamp extractors are Kafka records (`timestamp.extractor=Record`) or record fields (`timestamp.extractor=RecordField`). The deterministic rotation strategy configuration is `rotate.interval.ms` (setting `rotate.schedule.interval.ms` is nondeterministic and will invalidate exactly-once guarantees).