Awesome

Spark ↔ ElasticSearch

ElasticSearch integration for Apache Spark.

Features

• Transport client
• Partition per shard
• Data co-location
• Flexible and simple API

Usage

Add spark-es dependency to an SBT configuration file:

libraryDependencies += "com.github.shse" %% "spark-es" % "1.0.7"

Read from ElasticSearch using query_string query:

import org.apache.spark.elasticsearch._

val query = "name:john"

val documents = sparkContext.esRDD(Seq("localhost"), "cluster1", Seq("index1"), Seq("type1"), query)

Read from ElasticSearch using org.elasticsearch.action.search.SearchRequestBuilder:

import org.apache.spark.elasticsearch._

def getQuery = QueryBuilders.termQuery("name", "john") // Define query as a function to avoid serialization issues

val documents = 
  sparkContext.esRDD(Seq("localhost"), "SparkES", Seq("index1"), Seq("type1"), _.setQuery(getQuery))

Save to ElasticSearch:

import org.apache.spark.elasticsearch._

val documents = Seq(
  ESDocument(ESMetadata("1", "type1", "index1"), """{"name": "Sergey Shumov"}"""),
  ESDocument(ESMetadata("2", "type1", "index1"), """{"name": "John Smith"}""")
)

val options = SaveOptions(
  saveOperation = SaveOperation.Create, // Do not overwrite existing documents
  ignoreConflicts = true // Do not fail if document already exists
)

sparkContext
  .parallelize(documents, 2)
  .saveToES(Seq("localhost"), "cluster1", options)

Delete from ElasticSearch:

import org.apache.spark.elasticsearch._

// Using metadata RDD
val metadata = Seq(
  ESMetadata("1", "type1", "index1"),
  ESMetadata("2", "type1", "index1")
)

sparkContext
  .parallelize(documents, 2)
  .deleteFromES(Seq("localhost"), "cluster1")

// Using document indices
val ids = Seq("1", "2")

sparkContext
  .parallelize(ids, 2)
  .deleteFromES(Seq("localhost"), "cluster1", "index1", "type1")

Custom bulk action:

import org.apache.spark.elasticsearch._

val items = Seq("type1" -> "1", "type2" -> "2")

def handleResponse(response: BulkItemResponse): Unit =
  if (response.isFailed)
    println(response.getFailure.getStatus)

def handleDocument(client: Client, bulk: BulkRequestBuilder, document: (String, String): Unit =
  bulk.add(client.prepareDelete("index1", document._1, document._2))

sparkContext
  .parallelize(items, 2)
  .bulkToES(Seq("localhost"), "cluster1", handleDocument, handleResponse)

Building

Assembly:

sbt assembly