Configuration

elasticsearch-hadoop behavior can be customized through the properties below, typically by setting them on the target job’s Hadoop Configuration. However some of them can be specified through other means depending on the library used (see the relevant section).

es.resource = twitter/tweet   # index 'twitter', type 'tweet'

Note that specifying multiple indices and/or types in the above resource settings are allowed only for reading. Specifying multiple indices for writes is only supported through the use of dynamic resource (described below). 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 they have the same structure and only with some libraries. Integrations that require a strongly typed mapping (such as a table like Hive or SparkSQL) are likely to fail.

For writing, elasticsearch-hadoop 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 Elasticsearch. 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):

{
    "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:

# 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.

When using dynamic/multi writes, one can also specify a formatting of the value returned by the field. Out of the box, elasticsearch-hadoop 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 SimpleDataFormat syntax, one can format and parse the date in a locale-sensitive manner.

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

# index the documents based on their date
es.resource.write = my-collection/{@timestamp|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 SimpleDateFormat 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).

Logstash users will find this pattern quite familiar.

To wit, here is an example:

# 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.

For example:

# 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.

For example:

# To exclude field company.income
es.read.field.exclude = company.income

For example, given the document:

{
  "foo": {
    "bar": [ "abc", "def" ]
  }
}

Use the following configuration:

# mapping foo.bar as an array
es.read.field.as.array.include = foo.bar

If your document contains multiple levels of arrays:

{
  "foo": {
    "bar": [[["abc", "def"], ["ghi", "jkl"]], [["mno", "pqr"], ["stu", "vwx"]]]
  }
}

Then specify the dimensionality of the array in the configuration:

# mapping foo.bar as a 3-level/dimensional array
es.read.field.as.array.include = foo.bar:3

One using the update or upsert operation, additional settings (that mirror the update API) are available:

For example:

# specifying 2 parameters, one extracting the value from field 'number', the other containing the value '123':
es.update.script.params = param1:number,param2:<123>

For example:

es.update.script.params.json = {"param1":1, "param2":2}

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

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

es.read.shard.preference = _only_nodes:abc*

User specified source filters were found [name,timestamp], but the connector is executing in a state where it has provided its own source filtering [name,timestamp,location.address]. Please clear the user specified source fields under the [es.read.source.filter] property to continue. Bailing out...

For example, here the user is setting the Max-Forwards HTTP header:

es.net.http.header.Max-Forwards = 10