<commit_before>[[configuration]]
== Configuration

{eh} behavior can be customized through the properties below, typically by setting them on the target job Hadoop `Configuration`. However some of them can be specified through other means depending on the library used (see the relevant section).

****
{eh} uses the same conventions and reasonable defaults as {es} so you can give it a try out without bothering with the configuration. Most of the time, these defaults are just fine for running a production cluster; if you are fine-tunning your cluster or wondering about the effect of certain configuration option, please _do ask_ for more information.
****

NOTE: All configuration properties start with the `es` prefix. The namespace `es.internal` is reserved by the library for its internal use and should _not_ be used by the user at any point.

[float]
=== Required settings

`es.resource`::
{es} resource location, where data is read _and_ written to. Requires the format `<index>/<type>` (relative to the {es} host/port (see <<cfg-network,below>>))).

[source,ini]
----
es.resource = twitter/tweet   # index 'twitter', type 'tweet'
----

`es.resource.read` (defaults to `es.resource`)::
{es} resource used for reading (but not writing) data. Useful when reading and writing data to different {es} indices within the _same_ job. Typically set automatically (expect for the {mr} module which requires manual configuration).

`es.resource.write`(defaults to `es.resource`)::
{es} resource used for writing (but not reading) data. Used typically for __dynamic resource__ writes or when writing and reading data to different {es} indices within the _same_ job. Typically set automatically (expect for the {mr} module which requires manual configuration).

Note that [multiple](https://www.elastic.co/guide/en/elasticsearch/guide/current/multi-index-multi-type.html) indices and/or types are allowed *only* for reading. Use `_all/types` to search `types` in all indices or `index/` to search
all types within `index`.
Do note that reading multiple indices/types typically works only when the have the same structure and only with some libraries. Integrations that require a strongly type mapping (such as a table like Hive or SparkSQL) are likely to fail.

[[cfg-multi-writes]]
[float]
===== Dynamic/multi resource writes

For writing, {eh} allows the target resource to be resolved at runtime by using patterns (by using the `{<field-name>}` format), resolved at runtime based on the data being streamed to {es}. That is, one can save documents to a certain `index` or `type` based on one or multiple fields resolved from the document about to be saved.

For example, assuming the following document set (described here in JSON for readability - feel free to translate this into the actual Java objects):

[source,json]
----
{
    "media_type":"game",
    "title":"Final Fantasy VI",
    "year":"1994"
},
{
    "media_type":"book",
    "title":"<NAME>",
    "year":"2010"
},
{
    "media_type":"music",
    "title":"Surfing With The Alien",
    "year":"1987"
}
----

to index each of them based on their `media_type` one would use the following pattern:

[source,ini]
----
# index the documents based on their type
es.resource.write = my-collection/{media_type}
----

which would result in `Final Fantasy VI` indexed under `my-collection/game`, `Harry Potter` under `my-collection/book` and `Surfing With The Alien` under `my-collection/music`.
For more information, please refer to the dedicated integration section.

IMPORTANT: Dynamic resources are supported _only_ for writing, for doing multi-index/types reads, use an appropriate {ref}/search-search.html[search query].

[[cfg-multi-writes-format]]
[float]
===== Formatting dynamic/multi resource writes

When using dynamic/multi writes, one can also specify a formatting of the value returned by the field. Out of the box, {eh} provides formatting for date/timestamp fields which is useful for automatically grouping time-based data (such as logs)
 within a certain time range under the same index. By using the Java http://docs.oracle.com/javase/7/docs/api/java/text/SimpleDateFormat.html[SimpleDataFormat] syntax, one can format and parse the date in a locale-sensitive manager.

For example assuming the data contains a +@timestamp+ field, one can group the documents in _daily_ indices using the following configuration:

[source,ini]
----
# index the documents based on their date
es.resource.write = my-collection/{@timestamp:YYYY.MM.dd} <1>
----

<1> +@timestamp+ field formatting - in this case +YYYY.MM.dd+

The same configuration property is used (+es.resource.write+) however, through the special +:+ characters a formatting pattern is specified.
Please refer to the http://docs.oracle.com/javase/7/docs/api/java/text/SimpleDateFormat.html[SimpleDataFormat] javadocs for more information on the syntax supported.
In this case +YYYY.MM.dd+ translates the date into the year (specified by four digits), month by 2 digits followed by the day by two digits (such as +2015.01.28+).

http://logstash.net/[Logstash] users will find this _pattern_ quite http://logstash.net/docs/latest/filters/date[familiar].

[float]
=== Essential settings

[[cfg-network]]
[float]
==== Network
`es.nodes` (default localhost)::
List of {es} nodes to connect to. When using {es} remotely, _do_ set this option. Note that the list does _not_ have to contain _every_ node inside the {es} cluster; these are discovered automatically by {eh} by default (see below). Each node can have its HTTP/REST port specified manually (e.g. `mynode:9600`).

`es.port` (default 9200)::
Default HTTP/REST port used for connecting to {es} - this setting is applied to the nodes in `es.nodes` that do not have any port specified.

added[2.2]
`es.nodes.path.prefix` (default empty)::
Prefix to add to _all_ requests made to {es}. Useful in environments where the cluster is proxied/routed under a certain path. For example, if the cluster is located at +someaddress:someport/custom/path/prefix+, one would set +es.nodes.path.prefix+ to +/custom/path/prefix+.

[float]
==== Querying
`es.query` (default none)::
Holds the query used for reading data from the specified `es.resource`. By default it is not set/empty, meaning the entire data under the specified index/type is returned.
`es.query` can have three forms:

uri query;;
using the form `?uri_query`, one can specify a {ref}/search-uri-request.html[query string]. Notice the leading `?`.

query dsl;;
using the form `query_dsl` - note the query dsl needs to start with `{` and end with `}` as mentioned {ref}/search-request-body.html[here]

external resource;;
if none of the two above do match, {eh} will try to interpret the parameter as a path within the HDFS file-system. If that is not the case, it will try to load the resource from the classpath or, if that fails, from the Hadoop `DistributedCache`. The resource should contain either a `uri query` or a `query dsl`.

To wit, here is an example:
[source,ini]
----
# uri (or parameter) query
es.query = ?q=costinl

# query dsl
es.query = { "query" : { "term" : { "user" : "costinl" } } }

# external resource
es.query = org/mypackage/myquery.json
----

In other words, `es.query` is flexible enough so that you can use whatever search api you prefer, either inline or by loading it from an external resource.

TIP: We recommend using query dsl externalized in a file, included within the job jar (and thus available on its classpath). This makes it easy
to identify, debug and organize your queries.
Through-out the documentation we use the uri query to save text and increase readability - real-life queries quickly become unwielding when used as uris.

[float]
==== Operation

`es.input.json` (default false)::
Whether the input is already in JSON format or not (the default). Please see the appropriate section of each
integration for more details about using JSON directly.

`es.write.operation` (default index)::
The write operation {eh} should peform - can be any of:
`index` (default);; new data is added while existing data (based on its id) is replaced (reindexed).
`create`;; adds new data - if the data already exists (based on its id), an exception is thrown.
`update`;; updates existing data (based on its id). If no data is found, an exception is thrown.
`upsert`;; known as _merge_ or insert if the data does not exist, updates if the data exists (based on its id).

[float]
[[cfg-mapping]]
==== Mapping (when writing to {es})

`es.mapping.id` (default none)::
The document field/property name containing the document id.

`es.mapping.parent` (default none)::
The document field/property name containing the document parent. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.version` (default none)::
The document field/property name containing the document version. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.version.type` (default depends on +es.mapping.version+)::
Indicates the {ref}/docs-index_.html#_version_types[type of versioning] used.
If +es.mapping.version+ is undefined (default), its value is unspecified. If +es.mapping.version+ is specified, its value becomes +external+.

`es.mapping.routing` (default none)::
The document field/property name containing the document routing. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.ttl` (default none)::
The document field/property name containing the document time-to-live. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.timestamp` (default none)::
The document field/property name containing the document timestamp. To specify a constant, use the `<CONSTANT>` format.

added[2.1]
`es.mapping.include` (default none)::
Field/property to be included in the document sent to {es}. Useful for _extracting_ the needed data from entities. The syntax is similar
to that of {es} {ref}/search-request-source-filtering.html[include/exclude].
Multiple values can be specified by using a comma. By default, no value is specified meaning all properties/fields are included.

added[2.1]
`es.mapping.exclude` (default none)::
Field/property to be excluded in the document sent to {es}. Useful for _eliminating_ unneeded data from entities. The syntax is similar
to that of {es} {ref}/search-request-source-filtering.html[include/exclude].
Multiple values can be specified by using a comma. By default, no value is specified meaning no properties/fields are excluded.

For example:
[source,ini]
----
# extracting the id from the field called 'uuid'
es.mapping.id = uuid

# specifying a parent with id '123'
es.mapping.parent = \<123>

# combine include / exclude for complete control
# include
es.mapping.include = u*, foo.*
# exclude
es.mapping.exclude = *.description
----

Using the configuration above, each entry will have only its top-level fields, starting with u and nested fields under +foo+ included
in the document with the exception of any nested field named +description+. Additionally the document parent will be +123+ while the
document id extracted from field +uuid+.


[float]
[[cfg-field-info]]
==== Field information (when reading from {es})

added[2.1]
`es.mapping.date.rich` (default true)::
Whether to create a _rich_ +Date+ like object for +Date+ fields in {es} or returned them as primitives (+String+ or +long+). By default this is
true. The actual object type is based on the library used; noteable exception being Map/Reduce which provides no built-in +Date+ object and as such
+LongWritable+ and +Text+ are returned regardless of this setting.

added[2.2]
`es.field.read.as.array.include` (default empty)::
Fields/properties that should be considered as arrays/lists. Since {es} can map one or multiple values to a field, {eh} cannot determine from the mapping
whether to instantiate one value or a array type (depending on the library type). When encountering multiple values, {eh} will automatically use 
the array/list type but in strict mapping scenarios (like Spark SQL) this might lead to an unexpected schema change.
The syntax is similar to that of {es} {ref}/search-request-source-filtering.html[include/exclude]. 
Multiple values can be specified by using a comma. By default, no value is specified meaning no properties/fields are included.

For example:
[source,ini]
----
# mapping nested.bar as an array
es.field.read.as.array.include = nested.bar
----

`es.field.read.as.array.exclude` (default empty)::
Fields/properties that should be considered as arrays/lists. Similar to `es.field.read.as.array.include` above. Multiple values can be specified by using a comma. 
By default, no value is specified meaning no properties/fields are excluded (and since none is included as indicated above), no field is treated as array
before-hand.

[float]
==== Metadata (when reading from {es})

+es.read.metadata+ (default false)::
Whether to include the document metadata (such as id and version) in the results or not (default).

+es.read.metadata.field+ (default _metadata)::
The field under which the metadata information is placed. When +es.read.metadata+ is set to true, the information is returned as a +Map+ under the specified field.

+es.read.metadata.version+ (default false)::
Whether to include the document version in the returned metadata. Applicable only if +es.read.metadata+ is enabled.


[float]
[[cfg-update]]
==== Update settings (when writing to {es})

One using the `update` or `upsert` operation, additional settings (that mirror the {ref}/docs-update.html[update] API) are available:

`es.update.script` (default none)::
Script used for updating the document.

`es.update.script.lang` (default none)::
Script language. By default, no value is specified applying the node configuration.

`es.update.script.params` (default none)::
Script parameters (if any). The document (currently read) field/property who's value is used. To specify a constant, use the `<CONSTANT>` format.
Multiple values can be specified through commas (`,`)

For example:
[source,ini]
----
# specifying 2 parameters, one extracting the value from field 'number', the other containing the value '123':
es.update.script.params = param1:number,param2:\<123>
----

`es.update.script.params.json`::
Script parameters specified in `raw`, JSON format. The specified value is passed as is, without any further processing or filtering. Typically used for migrating existing update scripts.

For example:
[source,ini]
----
es.update.script.params.json = {"param1":1, "param2":2}
----

`es.update.retry.on.conflict` (default 0)::
How many times an update to a document is retried in case of conflict. Useful in concurrent environments.

[float]
=== Advanced settings

[[configuration-options-index]]
[float]
==== Index

`es.index.auto.create` (default yes)::
Whether {eh} should create an index (if its missing) when writing data to {es} or fail.

`es.index.read.missing.as.empty` (default no)::
Whether {eh} will allow reading of non existing indices (and return an empty data set) or not (and throw an exception)

`es.field.read.empty.as.null` (default yes)::
Whether {eh} will treat empty fields as `null`. This settings is typically not needed (as {eh} already handles the
null case) but is enabled for making it easier to work with text fields that haven't been sanitized yet.

`es.field.read.validate.presence` (default warn)::
To help out spot possible mistakes when querying data from Hadoop (which results in incorrect data being returned), {eh} can perform validation spotting missing fields and potential typos. Possible values are :
`ignore`;; no validation is performed
`warn`;; a warning message is logged in case the validation fails
`strict`;; an exception is thrown, halting the job, if a field is missing

The default (`warn`) will log any typos to the console when the job starts:

[source,bash]
----
WARN main mr.EsInputFormat - Field(s) [naem, adress] not found
   in the Elasticsearch mapping specified; did you mean [name, location.address]?
----

[float]
==== Network

`es.nodes.discovery` (default true)::
Whether to discovery the nodes within the {es} cluster or only to use the ones given in `es.nodes` for metadata queries. Note that this setting only applies during start-up; afterwards when reading and writing, {eh} uses the target index shards (and their hosting nodes) unless +es.nodes.client.only+ is enabled.

`es.nodes.client.only` (default false)::
Whether to use {es} {ref}/modules-node.html[client nodes] (or _load-balancers_). When enabled, {eh} will route _all_ its requests (after nodes discovery, if enabled) through the _client_ nodes within the cluster. Note this typically significantly reduces the node parallelism and thus it is disabled by default. Enabling it also
disables `es.nodes.data.only` (since a client node is a non-data node).

added[2.1.2]
`es.nodes.data.only` (default true)::
Whether to use {es} {ref}/modules-node.html[data nodes] only. When enabled, {eh} will route _all_ its requests (after nodes discovery, if enabled) through the _data_ nodes within the cluster. The purpose of this configuration setting is to avoid overwhelming non-data nodes as these tend to be "smaller" nodes. This is enabled by default.

added[2.2]
`es.nodes.wan.only` (default false)::
Whether the connector is used against an {es} instance in a cloud/restricted environment over the WAN, such as Amazon Web Services. In this mode, the connector disables discovery and _only_ connects through the declared +es.nodes+ during all operations, including reads and writes.
Note that in this mode, performance is _highly_  affected.

added[2.2]
`es.http.timeout` (default 1m)::
Timeout for HTTP/REST connections to {es}.

`es.http.retries` (default 3)::
Number of retries for establishing a (broken) http connection. The retries are applied for each _conversation_ with an {es} node. Once the retries are depleted, the connection will automatically be re-reouted to the next
available {es} node (based on the declaration of `es.nodes`, followed by the discovered nodes - if enabled).

`es.scroll.keepalive` (default 10m)::
The maximum duration of result scrolls between query requests.

`es.scroll.size` (default 50)::
Number of results/items returned by each individual per request.

added[2.2]
`es.scroll.limit` (default -1)::
Number of _total_ results/items returned by each individual scroll. A negative value indicates that all documents that match should be returned. Do note that this applies per scroll which is typically bound to one of the job tasks.
Thus the total number of documents returned is `LIMIT * NUMBER_OF_SCROLLS (OR TASKS)`

`es.action.heart.beat.lead` (default 15s)::
The lead to task timeout before {eh} informs Hadoop the task is still running to prevent task restart.

added[2.1]
[float]
==== Basic Authentication

`es.net.http.auth.user`:: Basic Authentication user name
`es.net.http.auth.pass`:: Basic Authentication password

added[2.1]
[float]
==== SSL

`es.net.ssl` (default false):: Enable SSL

`es.net.ssl.keystore.location`:: key store (if used) location (typically a URL, without a prefix it is interpreted as a classpath entry)

`es.net.ssl.keystore.pass`:: key store password

`es.net.ssl.keystore.type` (default JKS):: key store type. PK12 is an common, alternative format

`es.net.ssl.truststore.location`:: trust store location (typically a URL, without a prefix it is interpreted as a classpath entry)

`es.net.ssl.truststore.pass`:: trust store password

`es.net.ssl.cert.allow.self.signed` (default false):: Whether or not to allow self signed certificates

`es.net.ssl.protocol`(default TLS):: SSL protocol to be used

[float]
==== Proxy

`es.net.proxy.http.host`:: Http proxy host name
`es.net.proxy.http.port`:: Http proxy port
`es.net.proxy.http.user`:: Http proxy user name
`es.net.proxy.http.pass`:: Http proxy password
`es.net.proxy.http.use.system.props`(default yes):: Whether the use the system Http proxy properties (namely `http.proxyHost` and `http.proxyPort`) or not

`es.net.proxy.socks.host`:: Http proxy host name
`es.net.proxy.socks.port`:: Http proxy port
`es.net.proxy.socks.user`:: Http proxy user name
`es.net.proxy.socks.pass`:: Http proxy password
`es.net.proxy.socks.use.system.props`(default yes):: Whether the use the system Socks proxy properties (namely `socksProxyHost` and `socksProxyHost`) or not

NOTE: {eh} allows proxy settings to be applied only to its connection using the setting above. Take extra care when there is already a JVM-wide proxy setting (typically through system properties) to avoid unexpected behavior.

[float]
[[configuration-serialization]]
==== Serialization

`es.batch.size.bytes` (default 1mb)::
Size (in bytes) for batch writes using {es} {ref}/docs-bulk.html[bulk] API. Note the bulk size is allocated _per task_ instance. Always multiply by the number of tasks within a Hadoop job to get the total bulk size at runtime hitting {es}.

`es.batch.size.entries` (default 1000)::
Size (in entries) for batch writes using {es} {ref}/docs-bulk.html[bulk] API - (0 disables it). Companion to `es.batch.size.bytes`, once one matches, the batch update is executed. Similar to the size, this setting is _per task_ instance; it gets multiplied at runtime by the total number of Hadoop tasks running.

`es.batch.write.refresh` (default true)::
Whether to invoke an {ref}/indices-refresh.html[index refresh] or not after a bulk update has been completed. Note this is called only after the entire write (meaning multiple bulk updates) have been executed.

`es.batch.write.retry.count` (default 3)::
Number of retries for a given batch in case {es} is overloaded and data is rejected. Note that only the rejected data is retried. If there is still data rejected after the retries have been performad, the Hadoop job is cancelled (and fails). A negative value indicates infinite retries; be careful in setting this value as it can have unwanted side effects.

`es.batch.write.retry.wait` (default 10s)::
Time to wait between batch write retries.

`es.ser.reader.value.class` (default _depends on the library used_)::
Name of the `ValueReader` implementation for converting JSON to objects. This is set by the framework depending on the library ({mr}, Cascading, Hive, Pig, etc...) used.

`es.ser.writer.value.class` (default _depends on the library used_)::
Name of the `ValueWriter` implementation for converting objects to JSON. This is set by the framework depending on the library ({mr}, Cascading, Hive, Pig, etc...) used.

[[configuration-runtime]]
== Hadoop Map/Reduce runtime options

When using {eh}, it is important to be aware of the following Hadoop configurations that can influence the way Map/Reduce tasks are executed and in return {eh}.

IMPORTANT: Unfortunately, these settings need to be setup *manually* *before* the job / script configuration. Since {eh} is called too late in the life-cycle, after the task has been tasks have been already dispatched and as such, cannot influence the execution anymore.

[float]
=== Speculative execution

[quote, Yahoo! developer network]
____
As most of the tasks in a job are coming to a close, http://developer.yahoo.com/hadoop/tutorial/module4.html#tolerance[speculative execution] will schedule redundant copies of the remaining tasks across several nodes which do not have other work to perform. Therefore, the same input can be processed multiple times in parallel, to exploit differences in machine capabilities.
____

In other words, speculative execution is an *optimization*, enabled by default, that allows Hadoop to create duplicates tasks of those which it considers hanged or slowed down. When doing data crunching or reading resources, having duplicate tasks is harmless and means at most a waste of computation resources; however when writing data to an external store, this can cause data corruption through duplicates or unnecessary updates.
Since the 'speculative execution' behavior can be triggered by external factors (such as network or CPU load which in turn cause false positive) even in stable environments (virtualized clusters are particularly prone to this) and has a direct impact on data, {eh} disables this optimization for data safety.

Speculative execution can be disabled for the map and reduce phase - we recommend disabling in both cases - by setting to `false` the following two properties:

`mapred.map.tasks.speculative.execution`
`mapred.reduce.tasks.speculative.execution`

One can either set the properties by name manually on the `Configuration`/`JobConf` client:

[source,java]
----
jobConf.setSpeculativeExecution(false);
// or
configuration.setBoolean("mapred.map.tasks.speculative.execution", false);
configuration.setBoolean("mapred.reduce.tasks.speculative.execution", false);
----

or by passing them as arguments to the command line:

[source,bash]
----
$ bin/hadoop jar -Dmapred.map.tasks.speculative.execution=false \
                 -Dmapred.reduce.tasks.speculative.execution=false <jar>
----

[[security]]
== Security

{eh} can work in secure environments and has support for authentication and authorization. However it is important to understand that {eh} per-se is a _connector_, that is bridges two different systems. So when talking about security,
it is important to understand to what system it applies: the connector can run within a secure Hadoop environment talking to a vanilla/non-secured {es} cluster. Or vice-versa, it can run within a non-secured Spark environment while talking securely to a {es} cluster. Of course, the opposite can happen as well; the connector running within a secure Hadoop environment and communicating with a secured {es} cluster or the most common use-case, running from an open 
Spark environment to a default, non-secured {es} install.
This enumeration of setups is actually on purpose, to illustrate that based on what piece of the environment is secured, its respective connector configuration needs to be adjusted. <commit_msg>[DOC] Extend section on speculative execution<commit_after>[[configuration]]
== Configuration

{eh} behavior can be customized through the properties below, typically by setting them on the target job Hadoop `Configuration`. However some of them can be specified through other means depending on the library used (see the relevant section).

****
{eh} uses the same conventions and reasonable defaults as {es} so you can give it a try out without bothering with the configuration. Most of the time, these defaults are just fine for running a production cluster; if you are fine-tunning your cluster or wondering about the effect of certain configuration option, please _do ask_ for more information.
****

NOTE: All configuration properties start with the `es` prefix. The namespace `es.internal` is reserved by the library for its internal use and should _not_ be used by the user at any point.

[float]
=== Required settings

`es.resource`::
{es} resource location, where data is read _and_ written to. Requires the format `<index>/<type>` (relative to the {es} host/port (see <<cfg-network,below>>))).

[source,ini]
----
es.resource = twitter/tweet   # index 'twitter', type 'tweet'
----

`es.resource.read` (defaults to `es.resource`)::
{es} resource used for reading (but not writing) data. Useful when reading and writing data to different {es} indices within the _same_ job. Typically set automatically (expect for the {mr} module which requires manual configuration).

`es.resource.write`(defaults to `es.resource`)::
{es} resource used for writing (but not reading) data. Used typically for __dynamic resource__ writes or when writing and reading data to different {es} indices within the _same_ job. Typically set automatically (expect for the {mr} module which requires manual configuration).

Note that [multiple](https://www.elastic.co/guide/en/elasticsearch/guide/current/multi-index-multi-type.html) indices and/or types are allowed *only* for reading. Use `_all/types` to search `types` in all indices or `index/` to search
all types within `index`.
Do note that reading multiple indices/types typically works only when the have the same structure and only with some libraries. Integrations that require a strongly type mapping (such as a table like Hive or SparkSQL) are likely to fail.

[[cfg-multi-writes]]
[float]
===== Dynamic/multi resource writes

For writing, {eh} allows the target resource to be resolved at runtime by using patterns (by using the `{<field-name>}` format), resolved at runtime based on the data being streamed to {es}. That is, one can save documents to a certain `index` or `type` based on one or multiple fields resolved from the document about to be saved.

For example, assuming the following document set (described here in JSON for readability - feel free to translate this into the actual Java objects):

[source,json]
----
{
    "media_type":"game",
    "title":"Final Fantasy VI",
    "year":"1994"
},
{
    "media_type":"book",
    "title":"Harry Potter",
    "year":"2010"
},
{
    "media_type":"music",
    "title":"Surfing With The Alien",
    "year":"1987"
}
----

to index each of them based on their `media_type` one would use the following pattern:

[source,ini]
----
# index the documents based on their type
es.resource.write = my-collection/{media_type}
----

which would result in `Final Fantasy VI` indexed under `my-collection/game`, `Harry Potter` under `my-collection/book` and `Surfing With The Alien` under `my-collection/music`.
For more information, please refer to the dedicated integration section.

IMPORTANT: Dynamic resources are supported _only_ for writing, for doing multi-index/types reads, use an appropriate {ref}/search-search.html[search query].

[[cfg-multi-writes-format]]
[float]
===== Formatting dynamic/multi resource writes

When using dynamic/multi writes, one can also specify a formatting of the value returned by the field. Out of the box, {eh} provides formatting for date/timestamp fields which is useful for automatically grouping time-based data (such as logs)
 within a certain time range under the same index. By using the Java http://docs.oracle.com/javase/7/docs/api/java/text/SimpleDateFormat.html[SimpleDataFormat] syntax, one can format and parse the date in a locale-sensitive manager.

For example assuming the data contains a +@timestamp+ field, one can group the documents in _daily_ indices using the following configuration:

[source,ini]
----
# index the documents based on their date
es.resource.write = my-collection/{@timestamp:YYYY.MM.dd} <1>
----

<1> +@timestamp+ field formatting - in this case +YYYY.MM.dd+

The same configuration property is used (+es.resource.write+) however, through the special +:+ characters a formatting pattern is specified.
Please refer to the http://docs.oracle.com/javase/7/docs/api/java/text/SimpleDateFormat.html[SimpleDataFormat] javadocs for more information on the syntax supported.
In this case +YYYY.MM.dd+ translates the date into the year (specified by four digits), month by 2 digits followed by the day by two digits (such as +2015.01.28+).

http://logstash.net/[Logstash] users will find this _pattern_ quite http://logstash.net/docs/latest/filters/date[familiar].

[float]
=== Essential settings

[[cfg-network]]
[float]
==== Network
`es.nodes` (default localhost)::
List of {es} nodes to connect to. When using {es} remotely, _do_ set this option. Note that the list does _not_ have to contain _every_ node inside the {es} cluster; these are discovered automatically by {eh} by default (see below). Each node can have its HTTP/REST port specified manually (e.g. `mynode:9600`).

`es.port` (default 9200)::
Default HTTP/REST port used for connecting to {es} - this setting is applied to the nodes in `es.nodes` that do not have any port specified.

added[2.2]
`es.nodes.path.prefix` (default empty)::
Prefix to add to _all_ requests made to {es}. Useful in environments where the cluster is proxied/routed under a certain path. For example, if the cluster is located at +someaddress:someport/custom/path/prefix+, one would set +es.nodes.path.prefix+ to +/custom/path/prefix+.

[float]
==== Querying
`es.query` (default none)::
Holds the query used for reading data from the specified `es.resource`. By default it is not set/empty, meaning the entire data under the specified index/type is returned.
`es.query` can have three forms:

uri query;;
using the form `?uri_query`, one can specify a {ref}/search-uri-request.html[query string]. Notice the leading `?`.

query dsl;;
using the form `query_dsl` - note the query dsl needs to start with `{` and end with `}` as mentioned {ref}/search-request-body.html[here]

external resource;;
if none of the two above do match, {eh} will try to interpret the parameter as a path within the HDFS file-system. If that is not the case, it will try to load the resource from the classpath or, if that fails, from the Hadoop `DistributedCache`. The resource should contain either a `uri query` or a `query dsl`.

To wit, here is an example:
[source,ini]
----
# uri (or parameter) query
es.query = ?q=costinl

# query dsl
es.query = { "query" : { "term" : { "user" : "costinl" } } }

# external resource
es.query = org/mypackage/myquery.json
----

In other words, `es.query` is flexible enough so that you can use whatever search api you prefer, either inline or by loading it from an external resource.

TIP: We recommend using query dsl externalized in a file, included within the job jar (and thus available on its classpath). This makes it easy
to identify, debug and organize your queries.
Through-out the documentation we use the uri query to save text and increase readability - real-life queries quickly become unwielding when used as uris.

[float]
==== Operation

`es.input.json` (default false)::
Whether the input is already in JSON format or not (the default). Please see the appropriate section of each
integration for more details about using JSON directly.

`es.write.operation` (default index)::
The write operation {eh} should peform - can be any of:
`index` (default);; new data is added while existing data (based on its id) is replaced (reindexed).
`create`;; adds new data - if the data already exists (based on its id), an exception is thrown.
`update`;; updates existing data (based on its id). If no data is found, an exception is thrown.
`upsert`;; known as _merge_ or insert if the data does not exist, updates if the data exists (based on its id).

[float]
[[cfg-mapping]]
==== Mapping (when writing to {es})

`es.mapping.id` (default none)::
The document field/property name containing the document id.

`es.mapping.parent` (default none)::
The document field/property name containing the document parent. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.version` (default none)::
The document field/property name containing the document version. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.version.type` (default depends on +es.mapping.version+)::
Indicates the {ref}/docs-index_.html#_version_types[type of versioning] used.
If +es.mapping.version+ is undefined (default), its value is unspecified. If +es.mapping.version+ is specified, its value becomes +external+.

`es.mapping.routing` (default none)::
The document field/property name containing the document routing. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.ttl` (default none)::
The document field/property name containing the document time-to-live. To specify a constant, use the `<CONSTANT>` format.

`es.mapping.timestamp` (default none)::
The document field/property name containing the document timestamp. To specify a constant, use the `<CONSTANT>` format.

added[2.1]
`es.mapping.include` (default none)::
Field/property to be included in the document sent to {es}. Useful for _extracting_ the needed data from entities. The syntax is similar
to that of {es} {ref}/search-request-source-filtering.html[include/exclude].
Multiple values can be specified by using a comma. By default, no value is specified meaning all properties/fields are included.

added[2.1]
`es.mapping.exclude` (default none)::
Field/property to be excluded in the document sent to {es}. Useful for _eliminating_ unneeded data from entities. The syntax is similar
to that of {es} {ref}/search-request-source-filtering.html[include/exclude].
Multiple values can be specified by using a comma. By default, no value is specified meaning no properties/fields are excluded.

For example:
[source,ini]
----
# extracting the id from the field called 'uuid'
es.mapping.id = uuid

# specifying a parent with id '123'
es.mapping.parent = \<123>

# combine include / exclude for complete control
# include
es.mapping.include = u*, foo.*
# exclude
es.mapping.exclude = *.description
----

Using the configuration above, each entry will have only its top-level fields, starting with u and nested fields under +foo+ included
in the document with the exception of any nested field named +description+. Additionally the document parent will be +123+ while the
document id extracted from field +uuid+.


[float]
[[cfg-field-info]]
==== Field information (when reading from {es})

added[2.1]
`es.mapping.date.rich` (default true)::
Whether to create a _rich_ +Date+ like object for +Date+ fields in {es} or returned them as primitives (+String+ or +long+). By default this is
true. The actual object type is based on the library used; noteable exception being Map/Reduce which provides no built-in +Date+ object and as such
+LongWritable+ and +Text+ are returned regardless of this setting.

added[2.2]
`es.field.read.as.array.include` (default empty)::
Fields/properties that should be considered as arrays/lists. Since {es} can map one or multiple values to a field, {eh} cannot determine from the mapping
whether to instantiate one value or a array type (depending on the library type). When encountering multiple values, {eh} will automatically use 
the array/list type but in strict mapping scenarios (like Spark SQL) this might lead to an unexpected schema change.
The syntax is similar to that of {es} {ref}/search-request-source-filtering.html[include/exclude]. 
Multiple values can be specified by using a comma. By default, no value is specified meaning no properties/fields are included.

For example:
[source,ini]
----
# mapping nested.bar as an array
es.field.read.as.array.include = nested.bar
----

`es.field.read.as.array.exclude` (default empty)::
Fields/properties that should be considered as arrays/lists. Similar to `es.field.read.as.array.include` above. Multiple values can be specified by using a comma. 
By default, no value is specified meaning no properties/fields are excluded (and since none is included as indicated above), no field is treated as array
before-hand.

[float]
==== Metadata (when reading from {es})

+es.read.metadata+ (default false)::
Whether to include the document metadata (such as id and version) in the results or not (default).

+es.read.metadata.field+ (default _metadata)::
The field under which the metadata information is placed. When +es.read.metadata+ is set to true, the information is returned as a +Map+ under the specified field.

+es.read.metadata.version+ (default false)::
Whether to include the document version in the returned metadata. Applicable only if +es.read.metadata+ is enabled.


[float]
[[cfg-update]]
==== Update settings (when writing to {es})

One using the `update` or `upsert` operation, additional settings (that mirror the {ref}/docs-update.html[update] API) are available:

`es.update.script` (default none)::
Script used for updating the document.

`es.update.script.lang` (default none)::
Script language. By default, no value is specified applying the node configuration.

`es.update.script.params` (default none)::
Script parameters (if any). The document (currently read) field/property who's value is used. To specify a constant, use the `<CONSTANT>` format.
Multiple values can be specified through commas (`,`)

For example:
[source,ini]
----
# specifying 2 parameters, one extracting the value from field 'number', the other containing the value '123':
es.update.script.params = param1:number,param2:\<123>
----

`es.update.script.params.json`::
Script parameters specified in `raw`, JSON format. The specified value is passed as is, without any further processing or filtering. Typically used for migrating existing update scripts.

For example:
[source,ini]
----
es.update.script.params.json = {"param1":1, "param2":2}
----

`es.update.retry.on.conflict` (default 0)::
How many times an update to a document is retried in case of conflict. Useful in concurrent environments.

[float]
=== Advanced settings

[[configuration-options-index]]
[float]
==== Index

`es.index.auto.create` (default yes)::
Whether {eh} should create an index (if its missing) when writing data to {es} or fail.

`es.index.read.missing.as.empty` (default no)::
Whether {eh} will allow reading of non existing indices (and return an empty data set) or not (and throw an exception)

`es.field.read.empty.as.null` (default yes)::
Whether {eh} will treat empty fields as `null`. This settings is typically not needed (as {eh} already handles the
null case) but is enabled for making it easier to work with text fields that haven't been sanitized yet.

`es.field.read.validate.presence` (default warn)::
To help out spot possible mistakes when querying data from Hadoop (which results in incorrect data being returned), {eh} can perform validation spotting missing fields and potential typos. Possible values are :
`ignore`;; no validation is performed
`warn`;; a warning message is logged in case the validation fails
`strict`;; an exception is thrown, halting the job, if a field is missing

The default (`warn`) will log any typos to the console when the job starts:

[source,bash]
----
WARN main mr.EsInputFormat - Field(s) [naem, adress] not found
   in the Elasticsearch mapping specified; did you mean [name, location.address]?
----

[float]
==== Network

`es.nodes.discovery` (default true)::
Whether to discovery the nodes within the {es} cluster or only to use the ones given in `es.nodes` for metadata queries. Note that this setting only applies during start-up; afterwards when reading and writing, {eh} uses the target index shards (and their hosting nodes) unless +es.nodes.client.only+ is enabled.

`es.nodes.client.only` (default false)::
Whether to use {es} {ref}/modules-node.html[client nodes] (or _load-balancers_). When enabled, {eh} will route _all_ its requests (after nodes discovery, if enabled) through the _client_ nodes within the cluster. Note this typically significantly reduces the node parallelism and thus it is disabled by default. Enabling it also
disables `es.nodes.data.only` (since a client node is a non-data node).

added[2.1.2]
`es.nodes.data.only` (default true)::
Whether to use {es} {ref}/modules-node.html[data nodes] only. When enabled, {eh} will route _all_ its requests (after nodes discovery, if enabled) through the _data_ nodes within the cluster. The purpose of this configuration setting is to avoid overwhelming non-data nodes as these tend to be "smaller" nodes. This is enabled by default.

added[2.2]
`es.nodes.wan.only` (default false)::
Whether the connector is used against an {es} instance in a cloud/restricted environment over the WAN, such as Amazon Web Services. In this mode, the connector disables discovery and _only_ connects through the declared +es.nodes+ during all operations, including reads and writes.
Note that in this mode, performance is _highly_  affected.

added[2.2]
`es.http.timeout` (default 1m)::
Timeout for HTTP/REST connections to {es}.

`es.http.retries` (default 3)::
Number of retries for establishing a (broken) http connection. The retries are applied for each _conversation_ with an {es} node. Once the retries are depleted, the connection will automatically be re-reouted to the next
available {es} node (based on the declaration of `es.nodes`, followed by the discovered nodes - if enabled).

`es.scroll.keepalive` (default 10m)::
The maximum duration of result scrolls between query requests.

`es.scroll.size` (default 50)::
Number of results/items returned by each individual per request.

added[2.2]
`es.scroll.limit` (default -1)::
Number of _total_ results/items returned by each individual scroll. A negative value indicates that all documents that match should be returned. Do note that this applies per scroll which is typically bound to one of the job tasks.
Thus the total number of documents returned is `LIMIT * NUMBER_OF_SCROLLS (OR TASKS)`

`es.action.heart.beat.lead` (default 15s)::
The lead to task timeout before {eh} informs Hadoop the task is still running to prevent task restart.

added[2.1]
[float]
==== Basic Authentication

`es.net.http.auth.user`:: Basic Authentication user name
`es.net.http.auth.pass`:: Basic Authentication password

added[2.1]
[float]
==== SSL

`es.net.ssl` (default false):: Enable SSL

`es.net.ssl.keystore.location`:: key store (if used) location (typically a URL, without a prefix it is interpreted as a classpath entry)

`es.net.ssl.keystore.pass`:: key store password

`es.net.ssl.keystore.type` (default JKS):: key store type. PK12 is an common, alternative format

`es.net.ssl.truststore.location`:: trust store location (typically a URL, without a prefix it is interpreted as a classpath entry)

`es.net.ssl.truststore.pass`:: trust store password

`es.net.ssl.cert.allow.self.signed` (default false):: Whether or not to allow self signed certificates

`es.net.ssl.protocol`(default TLS):: SSL protocol to be used

[float]
==== Proxy

`es.net.proxy.http.host`:: Http proxy host name
`es.net.proxy.http.port`:: Http proxy port
`es.net.proxy.http.user`:: Http proxy user name
`es.net.proxy.http.pass`:: Http proxy password
`es.net.proxy.http.use.system.props`(default yes):: Whether the use the system Http proxy properties (namely `http.proxyHost` and `http.proxyPort`) or not

`es.net.proxy.socks.host`:: Http proxy host name
`es.net.proxy.socks.port`:: Http proxy port
`es.net.proxy.socks.user`:: Http proxy user name
`es.net.proxy.socks.pass`:: Http proxy password
`es.net.proxy.socks.use.system.props`(default yes):: Whether the use the system Socks proxy properties (namely `socksProxyHost` and `socksProxyHost`) or not

NOTE: {eh} allows proxy settings to be applied only to its connection using the setting above. Take extra care when there is already a JVM-wide proxy setting (typically through system properties) to avoid unexpected behavior.

[float]
[[configuration-serialization]]
==== Serialization

`es.batch.size.bytes` (default 1mb)::
Size (in bytes) for batch writes using {es} {ref}/docs-bulk.html[bulk] API. Note the bulk size is allocated _per task_ instance. Always multiply by the number of tasks within a Hadoop job to get the total bulk size at runtime hitting {es}.

`es.batch.size.entries` (default 1000)::
Size (in entries) for batch writes using {es} {ref}/docs-bulk.html[bulk] API - (0 disables it). Companion to `es.batch.size.bytes`, once one matches, the batch update is executed. Similar to the size, this setting is _per task_ instance; it gets multiplied at runtime by the total number of Hadoop tasks running.

`es.batch.write.refresh` (default true)::
Whether to invoke an {ref}/indices-refresh.html[index refresh] or not after a bulk update has been completed. Note this is called only after the entire write (meaning multiple bulk updates) have been executed.

`es.batch.write.retry.count` (default 3)::
Number of retries for a given batch in case {es} is overloaded and data is rejected. Note that only the rejected data is retried. If there is still data rejected after the retries have been performad, the Hadoop job is cancelled (and fails). A negative value indicates infinite retries; be careful in setting this value as it can have unwanted side effects.

`es.batch.write.retry.wait` (default 10s)::
Time to wait between batch write retries.

`es.ser.reader.value.class` (default _depends on the library used_)::
Name of the `ValueReader` implementation for converting JSON to objects. This is set by the framework depending on the library ({mr}, Cascading, Hive, Pig, etc...) used.

`es.ser.writer.value.class` (default _depends on the library used_)::
Name of the `ValueWriter` implementation for converting objects to JSON. This is set by the framework depending on the library ({mr}, Cascading, Hive, Pig, etc...) used.

[[configuration-runtime]]
== Runtime options

When using {eh}, it is important to be aware of the following Hadoop configurations that can influence the way Map/Reduce tasks are executed and in return {eh}.

IMPORTANT: Unfortunately, these settings need to be setup *manually* *before* the job / script configuration. Since {eh} is called too late in the life-cycle, after the task has been tasks have been already dispatched and as such, cannot influence the execution anymore.

[float]
=== Speculative execution

[quote, Yahoo! developer network]
____
As most of the tasks in a job are coming to a close, http://developer.yahoo.com/hadoop/tutorial/module4.html#tolerance[speculative execution] will schedule redundant copies of the remaining tasks across several nodes which do not have other work to perform. Therefore, the same input can be processed multiple times in parallel, to exploit differences in machine capabilities.
____

In other words, speculative execution is an *optimization*, enabled by default, that allows Hadoop to create duplicates tasks of those which it considers hanged or slowed down. When doing data crunching or reading resources, having duplicate tasks is harmless and means at most a waste of computation resources; however when writing data to an external store, this can cause data corruption through duplicates or unnecessary updates.
Since the 'speculative execution' behavior can be triggered by external factors (such as network or CPU load which in turn cause false positive) even in stable environments (virtualized clusters are particularly prone to this) and has a direct impact on data, {eh} disables this optimization for data safety.

Please check your library setting and disable this feature. If you encounter more data then expected, double and triple check this setting.

[float]
==== Disabling Map/Reduce speculative execution

Speculative execution can be disabled for the map and reduce phase - we recommend disabling in both cases - by setting to `false` the following two properties:

`mapred.map.tasks.speculative.execution`
`mapred.reduce.tasks.speculative.execution`

One can either set the properties by name manually on the `Configuration`/`JobConf` client:

[source,java]
----
jobConf.setSpeculativeExecution(false);
// or
configuration.setBoolean("mapred.map.tasks.speculative.execution", false);
configuration.setBoolean("mapred.reduce.tasks.speculative.execution", false);
----

or by passing them as arguments to the command line:

[source,bash]
----
$ bin/hadoop jar -Dmapred.map.tasks.speculative.execution=false \
                 -Dmapred.reduce.tasks.speculative.execution=false <jar>
----

[float]
==== Hive speculative execution

Apache Hive has its own setting for speculative execution through namely `hive.mapred.reduce.tasks.speculative.execution`. It is enabled by default so do change it to `false` in your scripts:

[source,sql]
----
set hive.mapred.reduce.tasks.speculative.execution=false;
----

Note that while the setting has been deprecated in Hive 0.10 and one might get a warning, double check that the speculative execution is actually disabled.

[float]
==== Spark speculative execution

Out of the box, Spark has speculative execution disable. Double check this is the case through `spark.speculation` setting (`false` disabled it, `true` to enable it).


[[security]]
== Security

{eh} can work in secure environments and has support for authentication and authorization. However it is important to understand that {eh} per-se is a _connector_, that is bridges two different systems. So when talking about security,
it is important to understand to what system it applies: the connector can run within a secure Hadoop environment talking to a vanilla/non-secured {es} cluster. Or vice-versa, it can run within a non-secured Spark environment while talking securely to a {es} cluster. Of course, the opposite can happen as well; the connector running within a secure Hadoop environment and communicating with a secured {es} cluster or the most common use-case, running from an open 
Spark environment to a default, non-secured {es} install.
This enumeration of setups is actually on purpose, to illustrate that based on what piece of the environment is secured, its respective connector configuration needs to be adjusted. 
{'fraction_non_alphanumeric': 0.09284413416299894, 'fraction_numerical': 0.003851115495522841, 'mean_word_length': 4.655901287553648, 'pattern_counts': {'":': 22, '<': 36, '<?xml version=': 0, '>': 36, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4886716', 'n_tokens_mistral': 14527, 'n_tokens_neox': 14036, 'n_words': 6959}