Elasticsearch 5.x Cookbook: Distributed Search and Analytics, Third Edition

To better understand the following sections, knowledge of the basic concepts such as application node and cluster are required.

One or more Elasticsearch nodes can be setup on physical or a virtual server depending on the available resources such as RAM, CPUs, and disk space.

A default node allows us to store data in it and to process requests and responses. (In Chapter 2, Downloading and Setup, we will see details on how to set up different nodes and cluster topologies).

When a node is started, several actions take place during its startup: such as:

After node startup, the node searches for other cluster members and checks its index and shard status.

To join two or more nodes in a cluster, these rules must be matched:

The network must be configured to support broadcast discovery (default) and they can communicate with each other. (Refer to How to setup networking recipe Chapter 2, Downloading and Setup).

A common approach in cluster management is to have one or more master nodes, which is the main reference for all cluster-level actions, and the other ones called secondary, that replicate the master data and actions.

To be consistent in write operations, all the update actions are first committed in the master node and then replicated in secondary ones.

In a cluster with multiple nodes, if a master node dies, a master-eligible one is elected to be the new master. This approach allows automatic failover to be setup in an Elasticsearch cluster.

In Elasticsearch, we have four kinds of nodes:

In big cluster architectures, having some nodes as simple client nodes with a lot of RAM, with no data, reduces the resources required by data nodes and improves performance in search using the local memory cache of them.

Starting an Elasticsearch node, a lot of output will be prompted; this output is provided during services start up. Every Elasticsearch server, that is running, provides services.

Elasticsearch natively provides a large set of functionalities that can be extended with additional plugins.

During a node startup, a lot of required services are automatically started. The most important ones are:

To work with Elasticsearch data, a user must have basic knowledge of data management and JSON (https://en.wikipedia.org/wiki/JSON) data format that is the lingua franca for working with Elasticsearch data and services.

Our main data container is called index (plural indices) and it can be considered similar to a database in the traditional SQL world. In an index, the data is grouped in data types called mappings in Elasticsearch. A mapping describes how the records are composed (fields). Every record, that must be stored in Elasticsearch, must be a JSON object.

Natively, Elasticsearch is a schema-less data store: when you put records in it, during insert it processes the records, splits it in fields, and updates the schema to manage the inserted data.

To manage huge volumes of records, Elasticsearch uses the common approach to split an index into multiple parts (shards) so that they can be spread on several nodes. The shard management is transparent to user usage: all common record operations are managed automatically in Elasticsearch's application layer.

Every record is stored in only a shard; the sharding algorithm is based on record ID, so many operations, that require loading and changing of records/objects, can be achieved without hitting all the shards, but only the shard (and their replicas) that contains your object.

The following schema compares Elasticsearch structure with SQL and MongoDB ones:

The following screenshot is a conceptual representation of an Elasticsearch cluster with three nodes, one index with four shards and replica set to 1 (primary shards are in bold):

Elasticsearch, to ensure safe operations on index/mapping/objects, internally has rigid rules about how to execute operations.

In Elasticsearch the operations are divided into:

When a record is saved in Elasticsearch, the destination shard is chosen based on:

Splitting an index in a shard allows you to store your data in different nodes, because Elasticsearch tries to balance the shard distribution on all the available nodes.

Every shard can contain up to 2³² records (about 4.9 Billions), so the real limit to shard size it is the storage size.

Shards contain your data, and during the search process all the shards are used to calculate and retrieve results: so Elasticsearch performance in big data scales horizontally with the number of shards.

All native records operations (that is, index, search, update, and delete) are managed in shards.

The shard management is completely transparent to the user. Only advanced users tend to change the default shard routing and management to cover their custom scenarios, for example, if there is a requirement to put customer data in the same shard to speed up his operations (search/index/analytics).

You need one or more nodes running to have a cluster. To test an effective cluster, you need at least two nodes (that can be on the same machine).

An index can have one or more replicas (full copies of your data, automatically managed by Elasticsearch): the shards are called primary ones if they are part of the primary replica, and secondary ones if they are part of other replicas.

To maintain consistency in write operations, the following workflow is executed:

During search operations, if there are some replicas, a valid set of shards is chosen randomly between primary and secondary to improve performances. Elasticsearch has several allocation algorithms to better distribute shards on nodes. For reliability, replicas are allocated in a way that if a single node becomes unavailable, there is always at least one replica of each shard that is still available on the remaining nodes.

The following figure shows some example of possible shards and replica configuration:

The replica has a cost to increase the indexing time due to data node synchronization and also the time spent to propagate the message to the slaves (mainly in an asynchronous way).

Related to the concept of replication, there is the cluster status indicator of the health of your cluster.

It can cover three different states:

The standard installation of Elasticsearch provides access via its web services on port 9200 for HTTP and 9300 for native Elasticsearch protocol. Simply starting an Elasticsearch server, you can communicate on these ports with it.

Elasticsearch is designed to be used as a RESTful server, so the main protocol is the HTTP usually on port 9200 and above. This is the only protocol that can be used by programming languages that don't run on a Java Virtual Machine (JVM).

Every protocol has advantages and disadvantages. It's important to choose the correct one depending on the kind of applications you are developing. If you are in doubt, choose the HTTP protocol layer that is the most standard and easy to use.

Choosing the right protocol depends on several factors, mainly architectural and performance related. This schema factorizes the advantages and disadvantages related to them:

You need a working Elasticsearch cluster. Using the default configuration, Elasticsearch enables the 9200 port on your server to communicate in HTTP.

The standard RESTful protocol is easy to integrate because it is the lingua franca for the Web and can be used by every programming language.

Now, I'll show how easy it is to fetch the Elasticsearch greeting API on a running server at 9200 port using several programming languages.

In Bash or Windows prompt, the request will be:

 curl -XGET http://127.0.0.1:9200

In Python, the request will be:

  import urllib
  result = urllib.open("http://127.0.0.1:9200")

In Java, the request will be:

import java.io.BufferedReader; 
import java.io.InputStream; 
import java.io.InputStreamReader; 
import java.net.URL; 
 
... 
try {             
// get URL content 
  URL url = new URL("http://127.0.0.1:9200");              
  URLConnection conn = url.openConnection();
// open the stream and put it into BufferedReader              
  BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream()));               
 
String inputLine;              
while ((inputLine = br.readLine()) != null){ 
System.out.println(inputLine);              
}              
br.close();               
System.out.println("Done");           
} catch (MalformedURLException e) {             
e.printStackTrace();          
} catch (IOException e) {              
e.printStackTrace();          
}

In Scala, the request will be:

scala.io.Source.fromURL("http://127.0.0.1:9200", "utf-8").getLines.mkString("\n") 

For every language sample, the response will be the same:

{
 "name" : "elasticsearch",
 "cluster_name" : "elasticsearch",
 "cluster_uuid" : "rbCPXgcwSM6CjnX8u3oRMA",
 "version" : {
 "number" : "5.1.1",
 "build_hash" : "5395e21",
 "build_date" : "2016-12-06T12:36:15.409Z",
 "build_snapshot" : false,
 "lucene_version" : "6.3.0"
 },
 "tagline" : "You Know, for Search"
}

Every client creates a connection to the server index / and fetches the answer. The answer is a JSON object.

You can call Elasticsearch server from any programming language that you like. The main advantages of this protocol are:

In this book, a lot of examples are done calling the HTTP API via the command-line curl program. This approach is very fast and allows you to test functionalities very quickly.

Every language provides drivers to best integrate Elasticsearch or RESTful web services. The Elasticsearch community provides official drivers that support the most used programming languages.

You need a working Elasticsearch cluster--the standard port for native protocol is 9300.

The steps required to use the native protocol in a Java environment are as follows (in Chapter 14, Java Integration we'll discuss it in detail):

To initialize a native client, some settings are required to properly configure it. The important ones are:

With these settings, it's possible to initialize a new client giving an IP address and port (default 9300).

This is the internal protocol used in Elasticsearch: it's the fastest protocol available to talk with Elasticsearch.

The native protocol is an optimized binary one and works only for JVM languages. To use this protocol, you need to include elasticsearch.jar in your JVM project. Because it depends on Elasticsearch implementation, it must be the same version of the Elasticsearch cluster.

To use this protocol, you also need to study the internals of Elasticsearch, so it's not so easy to use as HTTP protocol.

Native protocol is very useful for massive data import. But as Elasticsearch is mainly thought as a REST HTTP server to communicate with, it lacks support for everything is not standard in Elasticsearch core, such as plugins entry points. Using this protocol, you are unable to call entry points made by external plugins in an easy way.

The native protocol is the most used in the Java world and it will be deeply discussed in Chapters 14, Java Integration, Chapters 15, Scala Integration, and Chapter 17, Plugin Development.

For further details on Elasticsearch Java API, they are available on Elasticsearch site at https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/index.html.