Apache Hive integration

Hive abstracts Hadoop by abstracting it through SQL-like language, called HiveQL so that users can apply data defining and manipulating operations to it, just like with SQL. In Hive data sets are defined through tables (that expose type information) in which data can be loaded, selected and transformed through built-in operators or custom/user defined functions (or UDFs).

Make elasticsearch-hadoop jar available in the Hive classpath. Depending on your options, there are various ways to achieve that. Use ADD command to add files, jars (what we want) or archives to the classpath:

ADD JAR /path/elasticsearch-hadoop.jar;

As an alternative, one can use the command-line:

CLI configuration.

$ bin/hive --auxpath=/path/elasticsearch-hadoop.jar

or use the hive.aux.jars.path property specified either through the command-line or, if available, through the hive-site.xml file, to register additional jars (that accepts an URI as well):

$ bin/hive -hiveconf hive.aux.jars.path=/path/elasticsearch-hadoop.jar

or if the hive-site.xml configuration can be modified, one can register additional jars through the hive.aux.jars.path option (that accepts an URI as well):

hive-site.xml configuration.

<property>
  <name>hive.aux.jars.path</name>
  <value>/path/elasticsearch-hadoop.jar</value>
  <description>A comma separated list (with no spaces) of the jar files</description>
</property>

When using Hive, one can use TBLPROPERTIES to specify the configuration properties (as an alternative to Hadoop Configuration object) when declaring the external table backed by Elasticsearch:

CREATE EXTERNAL TABLE artists (...)
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = 'radio/artists',
              'es.index.auto.create' = 'false'); 

By default, elasticsearch-hadoop uses the Hive table schema to map the data in Elasticsearch, using both the field names and types in the process. There are cases however when the names in Hive cannot be used with Elasticsearch (the field name can contain characters accepted by Elasticsearch but not by Hive). For such cases, one can use the es.mapping.names setting which accepts a comma-separated list of mapped names in the following format: Hive field name:Elasticsearch field name

To wit:

CREATE EXTERNAL TABLE artists (...)
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = 'radio/artists',
            'es.mapping.names' = 'date:@timestamp, url:url_123'); 

With elasticsearch-hadoop, Elasticsearch becomes just an external table in which data can be loaded or read from:

CREATE EXTERNAL TABLE artists (
    id      BIGINT,
    name    STRING,
    links   STRUCT<url:STRING, picture:STRING>)
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = 'radio/artists'); 

-- insert data to Elasticsearch from another table called 'source'
INSERT OVERWRITE TABLE artists
    SELECT NULL, s.name, named_struct('url', s.url, 'picture', s.picture)
                    FROM source s;

For cases where the id (or other metadata fields like ttl or timestamp) of the document needs to be specified, one can do so by setting the appropriate mapping, namely es.mapping.id. Following the previous example, to indicate to Elasticsearch to use the field id as the document id, update the table properties:

CREATE EXTERNAL TABLE artists (
    id      BIGINT,
    ...)
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.mapping.id' = 'id'...);

For cases where the job input data is already in JSON, elasticsearch-hadoop allows direct indexing without applying any transformation; the data is taken as is and sent directly to Elasticsearch. In such cases, one needs to indicate the json input by setting the es.input.json parameter. As such, in this case elasticsearch-hadoop expects the output table to contain only one field, who’s content is used as the JSON document. That is, the library will recognize specific textual types (such as string or binary) or simply call (toString).

CREATE EXTERNAL TABLE json (data STRING) 
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = '...',
              'es.input.json` = 'yes'); 
...

One can index the data to a different resource, depending on the row being read, by using patterns. Coming back to the aforementioned media example, one could configure it as follows:

CREATE EXTERNAL TABLE media (
    name    STRING,
    type    STRING,
    year    STRING,
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = 'my-collection/{type}'); 

For each row about to be written, elasticsearch-hadoop will extract the type field and use its value to determine the target resource.

The functionality is also available when dealing with raw JSON - in this case, the value will be extracted from the JSON document itself. Assuming the JSON source contains documents with the following structure:

{
    "media_type":"music",
    "title":"Surfing With The Alien",
    "year":"1987"
}

the table declaration can be as follows:

CREATE EXTERNAL TABLE json (data STRING) 
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = 'my-collection/{media_type}', 
              'es.input.json` = 'yes');

Reading from Elasticsearch is strikingly similar:

CREATE EXTERNAL TABLE artists (
    id      BIGINT,
    name    STRING,
    links   STRUCT<url:STRING, picture:STRING>)
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES('es.resource' = 'radio/artists', 
              'es.query' = '?q=me*');          

-- stream data from Elasticsearch
SELECT * FROM artists;

Hive provides various types for defining data and internally uses different implementations depending on the target environment (from JDK native types to binary-optimized ones). Elasticsearch integrates with all of them, including and Serde2 lazy and lazy binary:

It is worth mentioning that rich data types available only in Elasticsearch, such as GeoPoint or GeoShape are supported by converting their structure into the primitives available in the table above. For example, based on its storage a geo_point might be returned as a string or an array.