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.

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

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.

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 for help running the Confluent Platform and installing the SFTP connector package.

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:

    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:

    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.

    Caution

    You must include a double dash (--) between the connector name and your flag. For more information, see this post.

    confluent local services connect connector load CsvSchemaSftp --config sftp.properties
    

    Important

    Don’t use the Confluent CLI in production environments.

  4. Validate messages are sent to Kafka serialized with Avro.

    kafka-avro-console-consumer --topic sftp-testing-topic --from-beginning --bootstrap-server localhost:9092
    

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:

    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:

    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.

    Caution

    You must include a double dash (--) between the connector name and your flag. For more information, see this post.

    confluent local services connect connector load TsvSftp --config sftp.properties
    

    Important

    Don’t use the Confluent CLI 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:

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 section of the SFTP Source Connector Configuration Properties.

Configuration Properties

For connector-specific configuration properties, see SFTP Source Connector Configuration Properties.