elastic4s

For non-blocking streaming of data, elastic4s provides a Reactive Streams implementation built using Akka. To use this, you must add the elastic4s-streams module dependency.

This implementation supports two primary patterns:

1. Elastic Subscriber: Stream data from a publisher into Elasticsearch.
2. Elastic Publisher: Stream documents from Elasticsearch out to subscribers.

The entry point for elastic4s is an instance of ElasticClient. You create it by passing an HTTP library implementation (like JavaClient) to the ElasticClient companion object.

JavaClient is configured using ElasticProperties, which accepts a single string containing the protocol, host, and port. For multiple nodes, provide a comma-separated list of endpoints.

// Single node
val props = ElasticProperties("http://host1:9200")
val client = ElasticClient(JavaClient(props))

// Multiple nodes
val nodes = ElasticProperties("http://host1:9200,host2:9200,host3:9200")
val client = ElasticClient(JavaClient(nodes))

The Suggestions API returns similar-looking terms (suggestions) for a given text using a suggestor. Suggestions are included in a search query by adding a suggestions block.

Key concepts:

• Suggestion Types: You can use term, fuzzyCompletion, completion, or phrase suggestion types.
• Naming: Each suggestion must have a unique name. This name is used to identify the suggestion in the response.
• Retrieval: The preferred way to access results is to use the original suggestion object (e.g., the object created via termSuggestion) to look up the response from the search result. This avoids manual casting.
• Entries and Options: A suggestion response contains entries. Each entry corresponds to a term in the input text. The options array within an entry contains the actual suggested terms.

The primary entry point for executing requests in elastic4s is the ElasticClient class. It is responsible for executing requests (like SearchRequest) and returning responses (like SearchResponse).

ElasticClient does not handle HTTP directly; instead, it delegates HTTP functions to an implementation of the HttpClient typeclass. This allows you to swap the underlying transport layer (e.g., Akka HTTP, STTP, or the official Java client) while keeping the same high-level ElasticClient API.

The SearchIterator provides a simple way to iterate over all results in a search by implementing scala.collection.Iterator. It handles the complexity of re-requesting data via Elasticsearch scrolls automatically.

Key characteristics:

• It is a blocking implementation: it will block between requests to fetch the next batch.
• The search request must specify a keepAlive value for scrolling.
• It can return marshalled domain objects (using an implicit HitReader) or raw Elasticsearch Hit objects.

If you require a non-blocking solution, use the Reactive Streams implementation instead.

implicit val reader : HitReader[MyType] =  ...
val iterator = SearchIterator.iterate[MyType](client, search(index).matchAllQuery.keepAlive("1m").size(50))
iterator.foreach(println)

When indexing documents, you can use RefreshPolicy to control when newly indexed documents become available for search via the Search API. Note that this affects visibility, not data consistency or durability.

Available policies:

• RefreshPolicy.IMMEDIATE: Forces a refresh immediately after the index operation. The request blocks until the refresh is complete and documents are searchable. Warning: Using this under heavy load can cause contention by forcing Elasticsearch to refresh too frequently.
• RefreshPolicy.WAIT_FOR: The request blocks until the next scheduled refresh occurs. Documents become searchable after that refresh.
• RefreshPolicy.None: The request returns as soon as Elasticsearch acknowledges it. Documents only become searchable after the next scheduled refresh.