Scala elasticsearch

elasticsearch-hadoop provides native integration between Elasticsearch and Apache Spark, in the form of an (Resilient Distributed Dataset) (or Pair to be precise) that can read data from Elasticsearch. The is offered in two flavors: one for Scala (which returns the data as with Scala collections) and one for Java (which returns the data as containing collections).

Configurationedit

To configure elasticsearch-hadoop for Apache Spark, one can set the various properties described in the Configuration chapter in the object:

Command-lineFor those that want to set the properties through the command-line (either directly or by loading them from a file), note that Spark only accepts those that start with the "spark." prefix and will ignore the rest (and depending on the version a warning might be thrown). To work around this limitation, define the elasticsearch-hadoop properties by appending the prefix (thus they become ) and elasticsearch-hadoop will automatically resolve them:

Writing data to Elasticsearchedit

With elasticsearch-hadoop, any can be saved to Elasticsearch as long as its content can be translated into documents. In practice this means the type needs to be a (whether a Scala or a Java one), a or a Scala case class. When that is not the case, one can easily transform the data in Spark or plug-in their own custom .

Scalaedit

When using Scala, simply import the package which, through the pimp my library pattern, enriches the any API with methods:

As an alternative to the implicit import above, one can use elasticsearch-hadoop Spark support in Scala through in the package which acts as a utility class allowing explicit method invocations. Additionally instead of s (which are convenient but require one mapping per instance due to their difference in structure), use a case class :

For cases where the id (or other metadata fields like or ) of the document needs to be specified, one can do so by setting the appropriate mapping namely . Following the previous example, to indicate to Elasticsearch to use the field as the document id, update the configuration (it is also possible to set the property on the though due to its global effect it is discouraged):

Javaedit

Java users have a dedicated class that provides a similar functionality to , namely in the (a package similar to Spark’s Java API):

The code can be further simplified by using Java 5 static imports. Additionally, the (who’s mapping is dynamic due to its loose structure) can be replaced with a :

Setting the document id (or other metadata fields like or ) is similar to its Scala counterpart, though potentially a bit more verbose depending on whether you are using the JDK classes or some other utilities (like Guava):

Writing existing JSON to Elasticsearchedit

For cases where the data in the is already in JSON, elasticsearch-hadoop allows direct indexing without applying any transformation; the data is taken as is and sent directly to Elasticsearch. As such, in this case, elasticsearch-hadoop expects either an containing or byte arrays (/), assuming each entry represents a JSON document. If the does not have the proper signature, the methods cannot be applied (in Scala they will not be available).

Scalaedit

Javaedit

Writing to dynamic/multi-resourcesedit

For cases when the data being written to Elasticsearch needs to be indexed under different buckets (based on the data content) one can use the field which accepts a pattern that is resolved from the document content, at runtime. Following the aforementioned media example, one could configure it as follows:

Scalaedit

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

Javaedit

As expected, things in Java are strikingly similar:

Handling document metadataedit

Elasticsearch allows each document to have its own metadata. As explained above, through the various mapping options one can customize these parameters so that their values are extracted from their belonging document. Further more, one can even include/exclude what parts of the data are sent back to Elasticsearch. In Spark, elasticsearch-hadoop extends this functionality allowing metadata to be supplied outside the document itself through the use of pairs. In other words, for s containing a key-value tuple, the metadata can be extracted from the key and the value used as the document source.

The metadata is described through the Java enum within package which identifies its type - , , , etc…​ Thus an keys can be a containing the for each document and its associated values. If key is not of type , elasticsearch-hadoop will consider the object as representing the document id and use it accordingly. This sounds more complicated than it is, so let us see some examples.

Scalaedit

Pair s, or simply put s with the signature can take advantage of the methods that are available either through the implicit import of package or object. To manually specify the id for each document, simply pass in the (not of type ) in your :

When more than just the id needs to be specified, one should use a with keys of type :

Javaedit

In a similar fashion, on the Java side, provides methods that are applied to (the equivalent in Java of ). Thus to save documents based on their ids one can use:

When more than just the id needs to be specified, one can choose to use a populated with keys of type :

Reading data from Elasticsearchedit

For reading, one should define the Elasticsearch that streams data from Elasticsearch to Spark.

Scalaedit

Similar to writing, the package, enriches the API with methods:

The method can be overloaded to specify an additional query or even a configuration (overriding ):

The documents from Elasticsearch are returned, by default, as a containing as the first element the document id and the second element the actual document represented through Scala collections, namely one `Map[String, Any]`where the keys represent the field names and the value their respective values.

Javaedit

Java users have a dedicated that works the same as its Scala counterpart however the returned values (or second element) returns the documents as native, collections.

In a similar fashion one can use the overloaded methods to specify a query or pass a object for advanced configuration. Let us see how this looks, but this time around using Java static imports. Further more, let us discard the documents ids and retrieve only the values:

By using the API, one gets a hold of Spark’s dedicated which are better suited in Java environments than the base (due to its Scala signatures). Moreover, the dedicated returns Elasticsearch documents as proper Java collections so one does not have to deal with Scala collections (which is typically the case with s). This is particularly powerful when using Java 8, which we strongly recommend as its lambda expressions make collection processing extremely concise.

To wit, let us assume one wants to filter the documents from the and return only those that contain a value that contains (please ignore the fact one can and should do the filtering directly through Elasticsearch).

In versions prior to Java 8, the code would look something like this:

with Java 8, the filtering becomes a one liner:

Reading data in JSON formatedit

In 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 use the dedicated methods. In this case, the connector will return the documents content as it is received from Elasticsearch without any processing as an in Scala or in Java with the keys representing the document id and the value its actual content in JSON format.

Type conversionedit

elasticsearch-hadoop automatically converts Spark built-in types to Elasticsearch types (and back) as shown in the table below:

in addition, the following implied conversion applies for Java types:

The conversion is done as a best effort; built-in Java and Scala types are guaranteed to be properly converted, however there are no guarantees for user types whether in Java or Scala. As mentioned in the tables above, when a class is encountered in Scala or in Java, the converters will try to its content and save it as an . Note this works only for top-level user objects - if the user object has other user objects nested in, the conversion is likely to fail since the converter does not perform nested . This is done on purpose since the converter has to serialize and deserialize the data and user types introduce ambiguity due to data loss; this can be addressed through some type of mapping however that takes the project way too close to the realm of ORMs and arguably introduces too much complexity for little to no gain; thanks to the processing functionality in Spark and the plugability in elasticsearch-hadoop one can easily transform objects into other types, if needed with minimal effort and maximum control.

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

Spark Streaming is an extension on top of the core Spark functionality that allows near real time processing of stream data. Spark Streaming works around the idea of s, or Discretized Streams. operate by collecting newly arrived records into a small and executing it. This repeats every few seconds with a new in a process called microbatching. The api includes many of the same processing operations as the api, plus a few other streaming specific methods. elasticsearch-hadoop provides native integration with Spark Streaming as of version 5.0.

When using the elasticsearch-hadoop Spark Streaming support, Elasticsearch can be targeted as an output location to index data into from a Spark Streaming job in the same way that one might persist the results from an . Though, unlike s, you are unable to read data out of Elasticsearch using a due to the continuous nature of it.

Writing to Elasticsearchedit

Like s, any can be saved to Elasticsearch as long as its content can be translated into documents. In practice this means the type needs to be a (either a Scala or a Java one), a or a Scala case class. When that is not the case, one can easily transform the data in Spark or plug-in their own custom .

Scalaedit

When using Scala, simply import the package which, through the pimp my library pattern, enriches the API with methods:

As an alternative to the implicit import above, one can use elasticsearch-hadoop Spark Streaming support in Scala through in the package which acts as a utility class allowing explicit method invocations. Additionally instead of s (which are convenient but require one mapping per instance due to their difference in structure), use a case class :

For cases where the id (or other metadata fields like or ) of the document needs to be specified, one can do so by setting the appropriate mapping namely . Following the previous example, to indicate to Elasticsearch to use the field as the document id, update the configuration (it is also possible to set the property on the though due to its global effect it is discouraged):

Javaedit

Java users have a dedicated class that provides a similar functionality to , namely in the package (a package similar to Spark’s Java API):

The code can be further simplified by using Java 5 static imports. Additionally, the (who’s mapping is dynamic due to its loose structure) can be replaced with a :

Setting the document id (or other metadata fields like or ) is similar to its Scala counterpart, though potentially a bit more verbose depending on whether you are using the JDK classes or some other utilities (like Guava):

Writing Existing JSON to Elasticsearchedit

For cases where the data being streamed by the is already serialized as JSON, elasticsearch-hadoop allows direct indexing without applying any transformation; the data is taken as is and sent directly to Elasticsearch. As such, in this case, elasticsearch-hadoop expects either a containing or byte arrays (/), assuming each entry represents a JSON document. If the does not have the proper signature, the methods cannot be applied (in Scala they will not be available).

Scalaedit

Javaedit

Writing to dynamic/multi-resourcesedit

For cases when the data being written to Elasticsearch needs to be indexed under different buckets (based on the data content) one can use the field which accepts a pattern that is resolved from the document content, at runtime. Following the aforementioned media example, one could configure it as follows:

Scalaedit

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

Javaedit

As expected, things in Java are strikingly similar:

Handling document metadataedit

Elasticsearch allows each document to have its own metadata. As explained above, through the various mapping options one can customize these parameters so that their values are extracted from their belonging document. Further more, one can even include/exclude what parts of the data are sent back to Elasticsearch. In Spark, elasticsearch-hadoop extends this functionality allowing metadata to be supplied outside the document itself through the use of pairs.

This is no different in Spark Streaming. For s containing a key-value tuple, the metadata can be extracted from the key and the value used as the document source.

The metadata is described through the Java enum within package which identifies its type - , , , etc…​ Thus a 's keys can be a containing the for each document and its associated values. If the key is not of type , elasticsearch-hadoop will consider the object as representing the document id and use it accordingly. This sounds more complicated than it is, so let us see some examples.

Scalaedit

Pair s, or simply put s with the signature can take advantage of the methods that are available either through the implicit import of package or object. To manually specify the id for each document, simply pass in the (not of type ) in your :

When more than just the id needs to be specified, one should use a with keys of type :

Javaedit

In a similar fashion, on the Java side, provides methods that are applied to (the equivalent in Java of ).

This tends to involve a little more work due to the Java API’s limitations. For instance, you cannot create a directly from a queue of s. Instead, you must create a regular of objects and convert the into a . This sounds complex, but it’s a simple work around for a limitation of the API.

First, we’ll create a pair function, that takes a object in, and returns it right back to the framework:

Then we’ll apply the pair function to a of s to create a and save it:

When more than just the id needs to be specified, one can choose to use a populated with keys of type . We’ll use the same typing trick to repack the as a :

Spark Streaming Type Conversionedit

The elasticsearch-hadoop Spark Streaming support leverages the same type mapping as the regular Spark type mapping. The mappings are repeated here for consistency:

in addition, the following implied conversion applies for Java types:

Geo typesIt is worth re-mentioning that rich data types available only in Elasticsearch, such as or are supported by converting their structure into the primitives available in the table above. For example, based on its storage a might be returned as a or a .

On top of the core Spark support, elasticsearch-hadoop also provides integration with Spark SQL. In other words, Elasticsearch becomes a native source for Spark SQL so that data can be indexed and queried from Spark SQL transparently.

Supported Spark SQL versionsedit

Spark SQL while becoming a mature component, is still going through significant changes between releases. Spark SQL became a stable component in version 1.3, however it is not backwards compatible with the previous releases. Further more Spark 2.0 introduced significant changed which broke backwards compatibility, through the API. elasticsearch-hadoop supports both version Spark SQL 1.3-1.6 and Spark SQL 2.0 through two different jars: and support Spark SQL 1.3-1.6 (or higher) while supports Spark SQL 2.0. In other words, unless you are using Spark 2.0, use

Spark SQL support is available under package.

API differencesFrom the elasticsearch-hadoop user perspectives, the differences between Spark SQL 1.3-1.6 and Spark 2.0 are fairly consolidated. This document describes at length the differences which are briefly mentioned below:

As conceptually, a is a , the documentation below will focus on Spark SQL 1.3-1.6.

Writing (Spark SQL 1.3+) to Elasticsearchedit

With elasticsearch-hadoop, s (or any for that matter) can be indexed to Elasticsearch.

Scalaedit

In Scala, simply import package which enriches the given class with methods; while these have the same signature as the package, they are designed for implementations:

Javaedit

In a similar fashion, for Java usage the dedicated package provides similar functionality through the :

Again, with Java 5 static imports this can be further simplied to:

Writing existing JSON to Elasticsearchedit

When using Spark SQL, if the input data is in JSON format, simply convert it to a (in Spark SQL 1.3) or a (for Spark SQL 2.0) (as described in Spark documentation) through / methods.

Using pure SQL to read from Elasticsearchedit

Spark SQL 1.2 introduced a new API for reading from external data sources, which is supported by elasticsearch-hadoop simplifying the SQL configured needed for interacting with Elasticsearch. Further more, behind the scenes it understands the operations executed by Spark and thus can optimize the data and queries made (such as filtering or pruning), improving performance.

Data Sources in Spark SQLedit

When using Spark SQL, elasticsearch-hadoop allows access to Elasticsearch through method. In other words, to create a / backed by Elasticsearch in a declarative manner:

In Spark 1.4, one would use the following similar API calls:

In Spark 1.5, this can be further simplified to:

Whatever API is used, once created, the can be accessed freely to manipulate the data.

The sources declaration also allows specific options to be passed in, namely:

Both options are explained in the next section. To specify the options (including the generic elasticsearch-hadoop ones), one simply passes a to the aforementioned methods:

For example:

Do note that due to the SQL parser, the (among other common characters used for delimiting) is not allowed; the connector tries to work around it by append the prefix automatically however this works only for specifying the configuration options with only one (like above). Because of this, if properties with multiple are needed, one should use the or methods above and pass the properties as a .

Push-Down operationsedit

An important hidden feature of using elasticsearch-hadoop as a Spark is that the connector understand the operations performed within the /SQL and, by default, will translate them into the appropriate QueryDSL. In other words, the connector pushes down the operations directly at the source, where the data is efficiently filtered out so that only the required data is streamed back to Spark. This significantly increases the queries performance and minimizes the CPU, memory and I/O on both Spark and Elasticsearch clusters as only the needed data is returned (as oppose to returning the data in bulk only to be processed and discarded by Spark). Note the push down operations apply even when one specifies a query - the connector will enhance it according to the specified SQL.

As a side note, elasticsearch-hadoop supports all the `Filter`s available in Spark (1.3.0 and higher) while retaining backwards binary-compatibility with Spark 1.3.0, pushing down to full extent the SQL operations to Elasticsearch without any user interference.

Operators those have been optimized as pushdown filters:

To wit, consider the following Spark SQL:

or in pure SQL:

The connector translates the query into:

Further more, the pushdown filters can work on terms (the default) or can be configured to be strict and provide matches (work only on fields). Unless one manually specifies the mapping, it is highly recommended to leave the defaults as they are. This and other topics are discussed at length in the Elasticsearch Reference Documentation.

Note that , available since elasticsearch-hadoop 2.2 for Spark 1.6 or higher, allows filters that are already pushed down to Elasticsearch to be processed/evaluated by Spark as well (default) or not. Turning this feature off, especially when dealing with large data sizes speed things up. However one should pay attention to the semantics as turning this off, might return different results (depending on how the data is indexed, vs ). In general, when turning strict on, one can disable as well.

Data Sources as tablesedit

Available since Spark SQL 1.2, one can also access a data source by declaring it as a Spark temporary table (backed by elasticsearch-hadoop):

Once defined, the schema is picked up automatically. So one can issue queries, right away:

As elasticsearch-hadoop is aware of the queries being made, it can optimize the requests done to Elasticsearch. For example, given the following query:

it knows only the and fields are required (the first to be returned to the user, the second for Spark’s internal filtering) and thus will ask only for this data, making the queries quite efficient.

Reading s (Spark SQL 1.3) from Elasticsearchedit

As you might have guessed, one can define a backed by Elasticsearch documents. Or even better, have them backed by a query result, effectively creating dynamic, real-time views over your data.

Scalaedit

Through the package, methods are available on the API:

And just as with the Spark core support, additional parameters can be specified such as a query. This is quite a powerful concept as one can filter the data at the source (Elasticsearch) and use Spark only on the results:

Controlling the schemaIn some cases, especially when the index in Elasticsearch contains a lot of fields, it is desireable to create a that contains only a subset of them. While one can modify the (by working on its backing ) through the official Spark API or through dedicated queries, elasticsearch-hadoop allows the user to specify what fields to include and exclude from Elasticsearch when creating the .

Through and properties, one can indicate what fields to include or exclude from the index mapping. The syntax is similar to that of Elasticsearch include/exclude. Multiple values can be specified by using a comma. By default, no value is specified meaning all properties/fields are included and no properties/fields are excluded.

For example:

Javaedit

For Java users, a dedicated API exists through . It is strikingly similar to however it allows configuration options to be passed in through Java collections instead of Scala ones; other than that using the two is exactly the same:

Better yet, the can be backed by a query result:

Spark SQL Type conversionedit

elasticsearch-hadoop automatically converts Spark built-in types to Elasticsearch types (and back) as shown in the table below:

While Spark SQL s have an equivalent in both Scala and Java and thus the RDD conversion can apply, there are slightly different semantics - in particular with the types due to the way Spark SQL handles them:

Geo Types Conversion TableIn addition to the table above, for Spark SQL 1.3 or higher, elasticsearch-hadoop performs automatic schema detection for geo types, namely Elasticsearch and . Since each type allows multiple formats ( accepts latitude and longitude to be specified in 4 different ways, while allows a variety of types (currently 9)) and the mapping does not provide such information, elasticsearch-hadoop will sample the determined geo fields at startup and retrieve an arbitrary document that contains all the relevant fields; it will parse it and thus determine the necessary schema (so for example it can tell whether a is specified as a or as an ).

Released as an experimental feature in Spark 2.0, Spark Structured Streaming provides a unified streaming and batch interface built into the Spark SQL integration. As of elasticsearch-hadoop 6.0, we provide native functionality to index streaming data into Elasticsearch.

Supported Spark Structured Streaming versionsedit

Spark Structured Streaming is considered generally available as of Spark v2.2.0. As such, elasticsearch-hadoop support for Structured Streaming (available in elasticsearch-hadoop 6.0+) is only compatible with Spark versions 2.2.0 and onward. Similar to Spark SQL before it, Structured Streaming may be subject to significant changes between releases before its interfaces are considered stable.

Spark Structured Streaming support is available under the and packages. It shares a unified interface with Spark SQL in the form of the api. Clients can interact with streaming s in almost exactly the same way as regular batch s with only a few exceptions.

Writing Streaming (Spark SQL 2.0+) to Elasticsearchedit

With elasticsearch-hadoop, Stream-backed s can be indexed to Elasticsearch.

Scalaedit

In Scala, to save your streaming based s and s to Elasticsearch, simply configure the stream to write out using the format, like so:

Javaedit

In a similar fashion, the format is available for Java usage as well:

Writing existing JSON to Elasticsearchedit

When using Spark SQL, if the input data is in JSON format, simply convert it to a (for Spark SQL 2.0) (as described in Spark documentation) through the 's format.

Sink commit log in Spark Structured Streamingedit

Spark Structured Streaming advertises an end-to-end fault-tolerant exactly-once processing model that is made possible through the usage of offset checkpoints and maintaining commit logs for each streaming query. When executing a streaming query, most sources and sinks require you to specify a "checkpointLocation" in order to persist the state of your job. In the event of an interruption, launching a new streaming query with the same checkpoint location will recover the state of the job and pick up where it left off. We maintain a commit log for elasticsearch-hadoop’s Elasticsearch sink implementation in a special directory under the configured checkpoint location:

Each file in the commit log directory corresponds to a batch id that has been committed. The log implementation periodically compacts the logs down to avoid clutter. You can set the location for the log directory a number of ways:

Here is a list of configurations that affect the behavior of Elasticsearch’s commit log:

Spark Structured Streaming Type conversionedit

Structured Streaming uses the exact same type conversion rules as the Spark SQL integration.

elasticsearch-hadoop automatically converts Spark built-in types to Elasticsearch types as shown in the table below:

While Spark SQL s have an equivalent in both Scala and Java and thus the RDD conversion can apply, there are slightly different semantics - in particular with the types due to the way Spark SQL handles them:

Using the Map/Reduce layeredit

Another way of using Spark with Elasticsearch is through the Map/Reduce layer, that is by leveraging the dedicated in elasticsearch-hadoop. However, unless one is stuck on elasticsearch-hadoop 2.0, we strongly recommend using the native integration as it offers significantly better performance and flexibility.

Configurationedit

Through elasticsearch-hadoop, Spark can integrate with Elasticsearch through its dedicated , and in case of writing, through . These are described at length in the Map/Reduce chapter so please refer to that for an in-depth explanation.

In short, one needs to setup a basic Hadoop object with the target Elasticsearch cluster and index, potentially a query, and she’s good to go.

From Spark’s perspective, the only thing required is setting up serialization - Spark relies by default on Java serialization which is convenient but fairly inefficient. This is the reason why Hadoop itself introduced its own serialization mechanism and its own types - namely s. As such, and s are required to return which, out of the box, Spark does not understand. The good news is, one can easily enable a different serialization (Kryo) which handles the conversion automatically and also does this quite efficiently.

Or if you prefer Scala

Note that the Kryo serialization is used as a work-around for dealing with types; one can choose to convert the types directly (from to types) - which is fine however for getting started, the one liner above seems to be the most effective.

Reading data from Elasticsearchedit

To read data, simply pass in the class - since it supports both the and the Map/Reduce APIs, you are free to use either method on 's, (which we recommend for conciseness reasons) or . Which ever you chose, stick with it to avoid confusion and problems down the road.

Old () APIedit

The Scala version is below:

New () APIedit

As expected, the API version is strikingly similar - replace with and with . That’s about it.

The Scala version is below:

Thanks to its Map/Reduce layer, elasticsearch-hadoop can be used from PySpark as well to both read and write data to Elasticsearch. To wit, below is a snippet from the Spark documentation (make sure to switch to the Python snippet):

Also, the SQL loader can be used as well: