Apache Storm support

With Storm, one can compute, transform and filter data typically in a streaming scenario. As opposed to the rest of the libraries mentioned in this documentation, Apache Storm is a computational framework that is not tied to Map/Reduce itself however it does integrate with Hadoop, mainly through HDFS.

In order to use elasticsearch-hadoop, its jar needs to be available in Storm’s classpath. The Storm documentation covers this in detail but in short, one can either have the jar available on all Storm nodes or have elasticsearch-hadoop part of the jar being deployed (which we recommend). The latter approach allows isolation between the jobs and since the jar is self-contained, can be easily be moved across environments without additional setup making it much more robust.

The Storm integration supports the configuration options described in the Configuration chapter plus a few more that are specific to Storm, namely:

// tick every 2 seconds
builder.setBolt(...).addConfiguration(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 2)

The configuration can be set through Storm’s Config class, in which case they are available globally or, individually for each EsSpout and EsBolt instance. In general, it is recommended to set globally only the properties that apply to all the components and are unlikely to change:

Config conf = new Config();
conf.put("es.index.auto.create", "true");
StormSubmitter.submitTopology("myTopology", conf, topology);

For this reason, typically, one should use the per-component configuration model since it allows different configurations to be used within the same Storm topology:

Map conf = new HashMap();
conf.put("es.index.auto.create", "true");
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("esSpout", new EsSpout("index/type", conf));

Through elasticsearch-hadoop, Elasticsearch is exposed to Storm through a native Bolt, namely org.elasticsearch.storm.EsBolt that writes the Storm Tuples to Elasticsearch:

import org.elasticsearch.storm.EsBolt; 

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new RandomSentenceSpout(), 10);
builder.setBolt(
    "es-bolt",
    new EsBolt(
        "storm/docs"
    ),
    5)
    .shuffleGrouping("spout");

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. Thus assuming the documents contain a field called sentenceId which is unique and is suitable for an identifier, one can update the job configuration as follows:

Map conf = new HashMap();
conf.put("es.mapping.id", "sentenceId");
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("esSpout", new EsSpout("index/type", conf));

If the data passed to Storm is already in JSON format, EsBolt can pass it directly to Elasticsearch without any transformation; the data is taken as is and sent over the wire. In such cases, one needs to indicate the JSON input by setting the es.input.json parameter to true. Furthermore, the Bolt expects the receiving Tuple to contain only one value/field representing the JSON document. By default, common textual types are recognized, such as chararray or bytearray; otherwise it falls back to calling toString to get a hold of the JSON content.

String json1 = "{\"reason\" : \"business\",\"airport\" : \"SFO\"}";  
String json2 = "{\"participants\" : 5,\"airport\" : \"OTP\"}";

Map conf = new HashMap();
conf.put("es.input.json", "true"); 

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("json-spout", new StringSpout(Arrays.asList(json1, json2)); 
builder.setBolt("es-bolt", new EsBolt("storm/json-trips", conf)) 
                                    .shuffleGrouping("json-spout");

In cases where the data needs to be indexed based on its content, one can choose the target index based on a Tuple field. Reusing the aforementioned media example, one can partition the documents based on their type. Assuming the document tuple contains fields media_type, title and year one can index them as follows:

builder.setBolt("es-bolt",
    new EsBolt("my-collection-{media_type}/doc") 
).shuffleGrouping("spout");

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.

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 EsBolt with the configuration:

Map conf = new HashMap();
conf.put("es.input.json", "true"); 

builder.setBolt("es-bolt",
    new EsBolt(
        "my-collection-{media_type}-{year}/doc",
        conf) 
    ).shuffleGrouping("spout");

As you can expect, for reading data (typically executing queries) elasticsearch-hadoop offers a dedicated Spout through org.elasticsearch.storm.EsSpout which executes the query in Elasticsearch and streams the results back to Apache Storm:

import org.elasticsearch.storm.EsSpout; 

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(
    "es-spout",
    new EsSpout(      
        "storm/docs", 
        "?q=me*),     
    5);               
builder.setBolt("bolt", new PrinterBolt()).shuffleGrouping("es-spout");

Since Storm requires each Spout to declare its fields when creating a topology, by default EsSpout declares for its tuples a generic doc field containing the documents returned (one per tuple) from Elasticsearch. When dealing with structured data (documents sharing the same fields), one can configure the EsSpout to declare as fields the document properties effectively unwrapping the document as a Tuple. By setting up es.storm.spout.fields, EsSpout will use them indicate to the Storm topology the tuple content and extract them from the returned document.

For example if the Elasticsearch documents contain 3 fields: name, age and gender by setting es.storm.spout.fields to name, age, gender, instead of returning a tuple with one field (doc, containing the document), a tuple containing the three named fields (name, age and gender) will be returned instead.