= HDFS Component
:doctitle: HDFS
:shortname: hdfs
:artifactid: camel-hdfs
:description: Read and write from/to an HDFS filesystem using Hadoop 2.x.
:since: 2.14
:supportlevel: Stable
:component-header: Both producer and consumer are supported
//Manually maintained attributes
:camel-spring-boot-name: hdfs

*Since Camel {since}*

*{component-header}*

The HDFS component enables you to read and write messages from/to an
HDFS file system using Hadoop 2.x. HDFS is the distributed file system
at the heart of http://hadoop.apache.org[Hadoop].

Maven users will need to add the following dependency to their `pom.xml`
for this component:

[source,xml]
------------------------------------------------------------
<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-hdfs</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>
------------------------------------------------------------

----------------------------------------
hdfs://hostname[:port][/path][?options]
----------------------------------------

The path is treated in the following way:

1.  as a consumer, if it's a file, it just reads the file, otherwise if
it represents a directory it scans all the file under the path
satisfying the configured pattern. All the files under that directory
must be of the same type.
2.  as a producer, if at least one split strategy is defined, the path
is considered a directory and under that directory the producer creates
a different file per split named using the configured
UuidGenerator.

When consuming from hdfs then in normal mode, a file is split into
chunks, producing a message per chunk. You can configure the size of the
chunk using the chunkSize option. If you want to read from hdfs and
write to a regular file using the file component, then you can use the
fileMode=Append to append each of the chunks together.


// component-configure options: START

// component-configure options: END

// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
// component options: END

// endpoint options: START

// endpoint options: END

// component headers: START
include::partial$component-endpoint-headers.adoc[]
// component headers: END

== KeyType and ValueType

* NULL it means that the key or the value is absent
* BYTE for writing a byte, the java Byte class is mapped into a BYTE
* BYTES for writing a sequence of bytes. It maps the java ByteBuffer
class
* INT for writing java integer
* FLOAT for writing java float
* LONG for writing java long
* DOUBLE for writing java double
* TEXT for writing java strings

BYTES is also used with everything else, for example, in Camel a file is
sent around as an InputStream, int this case is written in a sequence
file or a map file as a sequence of bytes.

== Splitting Strategy

In the current version of Hadoop opening a file in append mode is
disabled since it's not very reliable. So, for the moment, it's only
possible to create new files. The Camel HDFS endpoint tries to solve
this problem in this way:

* If the split strategy option has been defined, the hdfs path will be
used as a directory and files will be created using the configured
UuidGenerator
* Every time a splitting condition is met, a new file is created. +
 The splitStrategy option is defined as a string with the following
syntax: splitStrategy=<ST>:<value>,<ST>:<value>,*

where <ST> can be:

* BYTES a new file is created, and the old is closed when the number of
written bytes is more than <value>
* MESSAGES a new file is created, and the old is closed when the number
of written messages is more than <value>
* IDLE a new file is created, and the old is closed when no writing
happened in the last <value> milliseconds

note that this strategy currently requires either setting an IDLE value
or setting the HdfsConstants.HDFS_CLOSE header to false to use the
BYTES/MESSAGES configuration...otherwise, the file will be closed with
each message

for example:

-----------------------------------------------------------------
hdfs://localhost/tmp/simple-file?splitStrategy=IDLE:1000,BYTES:5
-----------------------------------------------------------------

it means: a new file is created either when it has been idle for more
than 1 second or if more than 5 bytes have been written. So, running
`hadoop fs -ls /tmp/simple-file` you'll see that multiple files have
been created.

== Controlling to close file stream

When using the xref:hdfs-component.adoc[HDFS] producer *without* a split
strategy, then the file output stream is by default closed after the
write. However you may want to keep the stream open, and only explicitly
close the stream later. For that you can use the header
`HdfsConstants.HDFS_CLOSE` (value = `"CamelHdfsClose"`) to control this.
Setting this value to a boolean allows you to explicit control whether
the stream should be closed or not.

Notice this does not apply if you use a split strategy, as there are
various strategies that can control when the stream is closed.

== Using this component in OSGi

There are some quirks when running this component in an OSGi environment
related to the mechanism Hadoop 2.x uses to discover different
`org.apache.hadoop.fs.FileSystem` implementations. Hadoop 2.x uses
`java.util.ServiceLoader` which looks for
`/META-INF/services/org.apache.hadoop.fs.FileSystem` files defining
available filesystem types and implementations. These resources are not
available when running inside OSGi.

As with `camel-hdfs` component, the default configuration files need to
be visible from the bundle class loader. A typical way to deal with it
is to keep a copy of `core-default.xml` (and e.g., `hdfs-default.xml`)
in your bundle root.

=== Using this component with manually defined routes

There are two options:

1.  Package `/META-INF/services/org.apache.hadoop.fs.FileSystem`
resource with bundle that defines the routes. This resource should list
all the required Hadoop 2.x filesystem implementations.
2.  Provide boilerplate initialization code which populates internal,
static cache inside `org.apache.hadoop.fs.FileSystem` class:

[source,java]
----------------------------------------------------------------------------------------------------
org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
conf.setClass("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class, FileSystem.class);
conf.setClass("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class, FileSystem.class);
...
FileSystem.get("file:///", conf);
FileSystem.get("hdfs://localhost:9000/", conf);
...
----------------------------------------------------------------------------------------------------

=== Using this component with Blueprint container

Two options:

1.  Package `/META-INF/services/org.apache.hadoop.fs.FileSystem`
resource with bundle that contains blueprint definition.
2.  Add the following to the blueprint definition file:

[source,xml]
------------------------------------------------------------------------------------------------------
<bean id="hdfsOsgiHelper" class="org.apache.camel.component.hdfs.HdfsOsgiHelper">
   <argument>
      <map>
         <entry key="file:///" value="org.apache.hadoop.fs.LocalFileSystem"  />
         <entry key="hdfs://localhost:9000/" value="org.apache.hadoop.hdfs.DistributedFileSystem" />
         ...
      </map>
   </argument>
</bean>

<bean id="hdfs" class="org.apache.camel.component.hdfs.HdfsComponent" depends-on="hdfsOsgiHelper" />
------------------------------------------------------------------------------------------------------

This way Hadoop 2.x will have correct mapping of URI schemes to
filesystem implementations.

=== Using this component with a HighAvailability configuration

In a HA setup, there will be multiple nodes (_configured through the *namedNodes* parameter_).
The "hostname" and "port" portion of the endpoint uri will no longer have a _"host"_ meaning, but it will represent the name given to the cluster.

You can choose whatever name you want for the cluster (_the name should follow the [a-zA-Z0-9] convention_).
This name will be sanitized by replacing the _dirty_ characters with underscore. This is done so that a host name or ip could pottentialy be used, if it makes sense to you.

The cluster name will be mapped to the HA filesystem with a coresponding proxy, with failover, and the _works_.

[source,java]
------------------------------------------------------------------------------------------------------
from("hdfs://node1_and_2_cluster/dir1/dir2?namedNodes=node1.exemple.org:8020,node2.exemple.org:8020").routeId(...)
...
------------------------------------------------------------------------------------------------------


=== Using this component with Kerberos authentication

The kerberos config file is read when the camel component is created, not when the endpoint is created.
Because of this, the config file must be set at startup, with a call like:

[source,java]
------------------------------------------------------------------------------------------------------
static {
  HdfsComponent.setKerberosConfigFile("/etc/security/kerb5.conf");
}
------------------------------------------------------------------------------------------------------



include::spring-boot:partial$starter.adoc[]
