<a id="hdfs-config-options"></a>

# Configuration Reference for HDFS 2 Sink Connector for Confluent Platform

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

```properties
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
```

Connector-specific configuration properties are described below.

## HDFS

`hdfs.url`
: The HDFS connection URL. This configuration has the format of hdfs://hostname:port and specifies the HDFS to export data to. This property is deprecated and will be removed in future releases. Use `store.url` instead.
  <br/>
  * Type: string
  * Default: null
  * Importance: high

`hadoop.conf.dir`
: The Hadoop configuration directory.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`hadoop.home`
: The Hadoop home directory.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`logs.dir`
: Top level directory to store the write ahead logs. Supports `${topic}` in the value, which will
  be replaced by the actual topic name. Supports `${0}`, …, `${n}` in conjunction with
  `topic.capture.groups.regex`. For details, see `topic.capture.groups.regex` configuration
  documentation.
  <br/>
  * Type: string
  * Default: logs
  * Importance: high

## Security

`hdfs.authentication.kerberos`
: Configuration indicating whether HDFS is using Kerberos for authentication.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: high
  * Dependents: `connect.hdfs.principal`, `connect.hdfs.keytab`, `hdfs.namenode.principal`, `kerberos.ticket.renew.period.ms`

`connect.hdfs.principal`
: The principal name to load from the keytab for Kerberos authentication.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`connect.hdfs.keytab`
: The path to the keytab file for the HDFS connector principal. This keytab file should only be readable by the connector user.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`hdfs.namenode.principal`
: The Kerberos principal name for the HDFS Namenode. The format is
  `nn/_HOST@REALM.TLD`.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`kerberos.ticket.renew.period.ms`
: The period in milliseconds to renew the Kerberos ticket.
  <br/>
  * Type: long
  * Default: 3600000 (1 hour)
  * Importance: low

## Connector

`format.class`
: The format class to use when writing data to the store. Format classes
  implement the `io.confluent.connect.storage.format.Format` interface.
  <br/>
  * Type: class
  * Default: `io.confluent.connect.hdfs.avro.AvroFormat`
  * Importance: high
  <br/>
  These classes are available by default:
  <br/>
  - `io.confluent.connect.hdfs.avro.AvroFormat`
  - `io.confluent.connect.hdfs.json.JsonFormat`
  - `io.confluent.connect.hdfs.orc.OrcFormat`
  - `io.confluent.connect.hdfs.parquet.ParquetFormat`
  - `io.confluent.connect.hdfs.string.StringFormat`

`flush.size`
: Number of records written to store before invoking file commits.
  <br/>
  * Type: int
  * Importance: high
  <br/>
  #### IMPORTANT
  **Rotation strategy logic:** In the following rotation strategies, the logic to flush files to storage is triggered when a new record arrives, *after* the defined interval or scheduled interval time. Flushing files is also triggered periodically by the `offset.flush.interval.ms` setting defined in the Connect worker configuration. The `offset.flush.interval.ms` setting defaults to 60000 ms (60 seconds). If you enable the properties `rotate.interval.ms` or `rotate.schedule.interval.ms` and ingestion rate is low, you should set `offset.flush.interval.ms` to a smaller value so that records flush at the rotation interval (or close to the interval) .  Leaving the `offset.flush.interval.ms` set to the default 60 seconds may cause records to stay in an open file for longer than expected, if no new records get processed that trigger rotation.

`rotate.interval.ms`
: The time interval in milliseconds to invoke file commits. This configuration ensures that file commits are invoked every configured interval. This configuration is useful when data ingestion rate is low and the connector didn’t write enough messages to commit files. The default value -1 means that this feature is disabled.
  <br/>
  * Type: long
  * Default: -1
  * Importance: high

`rotate.schedule.interval.ms`
: The time interval in milliseconds to periodically invoke file commits. This configuration ensures that file commits are invoked every configured interval. Time of commit will be adjusted to 00:00 of selected timezone. Commit will be performed at scheduled time regardless previous commit time or number of messages. This configuration is useful when you have to commit your data based on current server time, like at the beginning of every hour. The default value -1 means that this feature is disabled.
  <br/>
  * Type: long
  * Default: -1
  * Importance: medium

The following Avro converter properties can be used in the connector
configuration:

`schemas.cache.config`
: The size of the schema cache used in the Avro converter.
  <br/>
  * Type: int
  * Default: 1000
  * Importance: low

`enhanced.avro.schema.support`
: Enable enhanced Avro schema support in the Avro Converter. When set to `true`, this property preserves Avro schema package information and Enums when going from Avro schema to Connect schema. This information is added back in when going from Connect schema to Avro schema.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: low

`connect.meta.data`
: Allow the Connect converter to add its metadata to the output schema.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: low
  <br/>
  The `connect.meta.data` property preserves the following Connect schema metadata when going from Connect schema to Avro schema. The following metadata is added back in when going from Avro schema to Connect schema.
  <br/>
  * doc
  * version
  * parameters
  * default value
  * name
  * type

For detailed information and configuration examples for Avro converters listed
above, see [Using Kafka Connect with Schema Registry](/platform/current/schema-registry/connect.html#avro).

`retry.backoff.ms`
: The retry backoff in milliseconds. This config is used to notify Connect to retry delivering a message batch or performing recovery in case of transient exceptions.
  <br/>
  * Type: long
  * Default: 5000 (5 seconds)
  * Importance: low

`shutdown.timeout.ms`
: Clean shutdown timeout. This makes sure that asynchronous Hive metastore updates are completed during connector shutdown.
  <br/>
  * Type: long
  * Default: 3000 (3 seconds)
  * Importance: medium

`filename.offset.zero.pad.width`
: Width to zero-pad offsets in store’s filenames if offsets are too short in order to provide fixed-width filenames that can be ordered by simple lexicographic sorting.
  <br/>
  * Type: int
  * Default: 10
  * Valid Values: [0,…]
  * Importance: low

`avro.codec`
: The Avro compression codec to be used for output files. Available values: null, deflate, snappy and bzip2 (Codec source is org.apache.avro.file.CodecFactory).
  <br/>
  * Type: string
  * Default: null
  * Valid Values: [null, deflate, snappy, bzip2]
  * Importance: low

`parquet.codec`
: The Parquet compression codec to be used for output files.
  <br/>
  * Type: string
  * Default: snappy
  * Valid Values: [none, snappy, gzip, brotli, lz4, lzo, zstd]
  * Importance: low

## Storage

`storage.class`
: The underlying storage layer.
  <br/>
  * Type: class
  * Default: io.confluent.connect.hdfs.storage.HdfsStorage
  * Importance: high

`topics.dir`
: Top level directory to store the data ingested from Apache Kafka®. Supports `${topic}` in the value,
  which will be replaced by the actual topic name. Supports `${0}`, …, `${n}` in conjunction
  with `topic.capture.groups.regex`. For details, see `topic.capture.groups.regex` configuration
  documentation.
  <br/>
  * Type: string
  * Default: topics
  * Importance: high

`store.url`
: Store’s connection URL, if applicable. For example: `hdfs://hostname:port`.
  <br/>
  * Type: string
  * Default: null
  * Importance: high

`directory.delim`
: Directory delimiter pattern
  <br/>
  * Type: string
  * Default: /
  * Importance: medium

`topic.capture.groups.regex`
: A Java Pattern regex that matches the entire topic and captures values for substituting into `topics.dir`. Indexed capture groups are accessible with `${n}`, where `${0}` refers to the whole match and `${1}` refers to the first capture group. Example config value of `([a-zA-Z]*)_([a-zA-Z]*)` will match topics that are two words delimited by an underscore and will capture each word separately. With `topic.dir = ${1}/${2}`, a record from the topic `example_name` will be written into a subdirectory of `example/name/`. By default, this functionality is not enabled.
  <br/>
  * Type: string
  * Default: null
  * Importance: low

## Partitioner

`partitioner.class`
: The partitioner to use when writing data to the store. The following partitioners are available:
  <br/>
  - `DefaultPartitioner` preserves the Kafka partitions.
  - `DailyPartitioner` partitions data according to date.
  - `HourlyPartitioner` partitions data according to hour.
  - `FieldPartitioner` partitions the data to different directories according to the value of the partitioning field
    specified in `partition.field.name`.
  - `TimeBasedPartitioner` partitions data according to ingestion time.
  <br/>
  * Type: class
  * Default: io.confluent.connect.storage.partitioner.DefaultPartitioner
  * Importance: high
  * Dependents: `partition.field.name`, `partition.duration.ms`, `path.format`, `locale`, `timezone`

`partition.field.name`
: The name of the partitioning field when `FieldPartitioner` is used. You can enter multiple partitioning field names using comma-separated names.
  <br/>
  * Type: list
  * Default: “”
  * Importance: medium

`partition.duration.ms`
: The duration of a partition milliseconds used by `TimeBasedPartitioner`. The default value -1 means that we are not using `TimeBasedPartitioner`.
  <br/>
  * Type: long
  * Default: -1
  * Importance: medium

`path.format`
: This configuration is used to set the format of the data directories when partitioning with `TimeBasedPartitioner`. The format set in this configuration converts the Unix timestamp to proper directories strings. For example, if you set `path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH`, the data directories will have the format `/year=2015/month=12/day=07/hour=15/`.
  <br/>
  * Type: string
  * Default: “”
  * Importance: medium

`locale`
: The locale to use when partitioning with `TimeBasedPartitioner`. Used to format dates and times. For example, use `en-US` for US English, `en-GB` for UK English, or `fr-FR` for French (in France). These may vary by Java version. See the [available locales](https://www.localeplanet.com/java/).
  <br/>
  * Type: string
  * Default: “”
  * Importance: medium

`timezone`
: The timezone to use when partitioning with `TimeBasedPartitioner`. Used to format and compute dates and times. All timezone IDs must be specified in the long format, such as `America/Los_Angeles`, `America/New_York`, and `Europe/Paris`, or `UTC`. Alternatively a locale independent, fixed offset, datetime zone can be specified in form `[+-]hh:mm`. Support for these timezones may vary by Java version. See the [available timezones within each locale](https://www.localeplanet.com/java/), such as [those within the US English locale](https://www.localeplanet.com/java/en-US/index.html).
  <br/>
  * Type: string
  * Default: “”
  * Valid Values: Any timezone accepted by: class `org.joda.time.DateTimeZone`
  * Importance: medium

`timestamp.extractor`
: The extractor that gets the timestamp for records when partitioning with `TimeBasedPartitioner`. It can be set to `Wallclock`, `Record` or `RecordField` in order to use one of the built-in timestamp extractors or be given the fully-qualified class name of a user-defined class that extends the `TimestampExtractor` interface.
  <br/>
  * Type: string
  * Default: Wallclock
  * Importance: medium

`timestamp.field`
: The record field to be used as timestamp by the timestamp extractor.
  <br/>
  * Type: string
  * Default: timestamp
  * Importance: medium

## Hive

`hive.integration`
: Configuration indicating whether to integrate with Hive when running the connector.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: high
  * Dependents: `hive.metastore.uris`, `hive.conf.dir`, `hive.home`, `hive.database`

`hive.metastore.uris`
: The Hive metastore URIs, can be IP address or fully-qualified domain name and port of the metastore host.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`hive.conf.dir`
: Hive configuration directory
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`hive.home`
: Hive home directory.
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`hive.database`
: The database to use when the connector creates tables in Hive.
  <br/>
  * Type: string
  * Default: default
  * Importance: high

`hive.table.name`
: The hive table name to use.
  It must contain `${topic}` to inject the corresponding topic name.
  <br/>
  * Type: string
  * Default: “${topic}”
  * Importance: low

## Schema

`schema.compatibility`
: The schema compatibility rule to use when the connector is observing schema changes. The supported configurations are NONE, BACKWARD, FORWARD and FULL.
  <br/>
  * Type: string
  * Default: NONE
  * Importance: high