Apache Pig support

It provides a high-level, powerful, scripting-like transformation language which gets compiled into Map/Reduce jobs at runtime by the Pig compiler. To simplify working with arbitrary data, Pig associates a schema (or type information) with each data set for validation and performance. This in turn, breaks it down into discrete data types that can be transformed through various operators or custom functions (or UDFs). Data can be loaded from and stored to various storages such as the local file-system or HDFS, and with elasticsearch-hadoop into Elasticsearch as well.

In order to use elasticsearch-hadoop, its jar needs to be in Pig’s classpath. There are various ways of making that happen though typically the REGISTER command is used:

REGISTER /path/elasticsearch-hadoop.jar;

As an alternative, when using the command-line, one can register additional jars through the -Dpig.additional.jars option (that accepts an URI as well):

$ pig -Dpig.additional.jars=/path/elasticsearch-hadoop.jar:<other.jars> script.pig

or if the jars are on HDFS

$ pig \
-Dpig.additional.jars=hdfs://<cluster-name>:<cluster-port>/<path>/elasticsearch-hadoop.jar:<other.jars> script.pig

With Pig, one can specify the configuration properties (as an alternative to Hadoop Configuration object) as a constructor parameter when declaring EsStorage:

STORE B INTO 'radio/artists' 
       USING org.elasticsearch.hadoop.pig.EsStorage
             ('es.http.timeout = 5m', 
              'es.index.auto.create = false'); 

DEFINE EsStorage org.elasticsearch.hadoop.pig.EsStorage();

Do note that it is possible (and recommended) to specify the configuration parameters to reduce script duplication, such as es.query or es.mapping.names:

DEFINE EsStorage org.elasticsearch.hadoop.pig.EsStorage('my.cfg.param=value');

Among the various types available in Pig, tuples are used the most. Tuples are defined as “ordered sets of fields” (e.g. (19,2)) however structurally they are shaped as ordered maps since each field has a name, which may be defined or not (e.g. (field:19, another:2)). The ordered aspect is important and forces elasticsearch-hadoop to use JSON arrays for tuples (using JSON objects is not an option as it does not preserve ordering besides the fact that it requires keys/names which might be or not available in a tuple). Obeying the rule of least surprise, elasticsearch-hadoop by default will disregard a tuple’s field names, both when writing and reading.

To change this behavior (which in effect means treating tuples as arrays of maps instead of arrays), use the boolean property es.mapping.pig.tuple.use.field.names (by default false) and set it to true.

The table below illustrates the difference between the two settings:

By default, Pig will only use one reducer per job which in most cases is inefficient. To address these issue:

-- launch the Map/Reduce job with 5 reducers
SET default_parallel 5;

or by using the PARALLEL keyword with COGROUP, CROSS, DISTINCT, GROUP, JOIN(inner), JOIN(outer) and ORDER BY.

B = GROUP A BY t PARALLEL 18;

20yy-mm-dd hh:mm:ss,mss [JobControl] INFO  org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil - Total input paths to process : 25
20yy-mm-dd hh:mm:ss,mss [JobControl] INFO  org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil - Total input paths (combined) to process : 1

Avoid this by setting pig.noSplitCombination to true (one can also use pig.splitCombination to false however we recommend the former) either by setting the property before invoking the script:

pig -Dpig.noSplitCombination=true myScript.pig

in the Pig script itself:

SET pig.noSplitCombination TRUE;

or through the global pig.properties configuration in your Pig install:

pig.noSplitCombination=true

Unfortunately elasticsearch-hadoop cannot set these properties automatically so the user has to do that manually per script or making them global through the Pig configuration as described above.

Out of the box, elasticsearch-hadoop uses the Pig 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 Pig cannot be used with Elasticsearch (invalid characters, existing names with different layout, etc…​). For such cases, one can use the es.mapping.names setting which accepts a comma-separated list of mapped names in the following format: Pig field name : Elasticsearch field name

For example:

STORE B INTO  '...' USING org.elasticsearch.hadoop.pig.EsStorage(
    'es.mapping.names=date:@timestamp, uRL:url')         

Elasticsearch is exposed as a native Storage to Pig so it can be used to store data into it:

-- load data from HDFS into Pig using a schema
A = LOAD 'src/test/resources/artists.dat' USING PigStorage()
                    AS (id:long, name, url:chararray, picture: chararray);
-- transform data
B = FOREACH A GENERATE name, TOTUPLE(url, picture) AS links;
-- save the result to Elasticsearch
STORE B INTO 'radio/artists'
       USING org.elasticsearch.hadoop.pig.EsStorage(); 

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 Storage configuration:

STORE B INTO 'radio/artists USING org.elasticsearch.hadoop.pig.EsStorage('es.mapping.id=id'...);

When 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 to receive a tuple with a single field (representing the JSON document); the library will recognize common textual types such as chararray or bytearray otherwise it just calls toString to get a hold of the JSON content.

A = LOAD '/resources/artists.json' USING PigStorage() AS (json:chararray);" 
STORE B INTO 'radio/artists'
    USING org.elasticsearch.hadoop.pig.EsStorage('es.input.json=true'...); 

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

A = LOAD 'src/test/resources/media.dat' USING PigStorage()
            AS (name:chararray, type:chararray, year: chararray); 
STORE B INTO 'my-collection-{type}/doc' 
       USING org.elasticsearch.hadoop.pig.EsStorage();

For each tuple 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":"game",
    "title":"Final Fantasy VI",
    "year":"1994"
}

the table declaration can be as follows:

A = LOAD '/resources/media.json' USING PigStorage() AS (json:chararray);" 
STORE B INTO 'my-collection-{media_type}/doc' 
    USING org.elasticsearch.hadoop.pig.EsStorage('es.input.json=true');

As you would expect, loading the data is straight forward:

-- execute Elasticsearch query and load data into Pig
A = LOAD 'radio/artists' 
    USING org.elasticsearch.hadoop.pig.EsStorage('es.query=?me*'); 
DUMP A;

In the case where the results from Elasticsearch need to be in JSON format (typically to be sent down the wire to some other system), one can instruct elasticsearch-hadoop to return the data as is. By setting es.output.json to true, the connector will parse the response from Elasticsearch, identify the documents and, without converting them, return their content to the user as String/chararray objects.

Pig internally uses native java types for most of its types and elasticsearch-hadoop abides to that convention.

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 chararray or a tuple.