ElasticSearch Cookbook

To better understand the upcoming sections, some knowledge of basic concepts of application node and cluster is required.

One or more ElasticSearch nodes can be set up on a physical or a virtual server depending on available resources such as RAM, CPUs, and disk space. A default node allows storing data in it and to process requests and responses. (In Chapter 2, Downloading and Setting Up ElasticSearch,we'll see details about how to set up different nodes and cluster topologies). When a node is started, several actions take place during its startup:

After node startup, the node searches for other cluster members and checks its indices and shards status. In order to join two or more nodes in a cluster, the following rules must be matched:

Refer to the Networking setup recipe in the next chapter.

A common approach in cluster management is to have a master node, which is the main reference for all cluster level actions, and the others ones called secondary or slaves, that replicate the master data and actions. 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 secondary one is elected to be the new master; this approach allows automatic failover to be set up in an ElasticSearch cluster.

There are two important behaviors in an ElasticSearch node, namely the arbiter and the data container.

The arbiter nodes are able to process the REST response and all the other operations of search. During every action execution, ElasticSearch generally executes actions using a MapReduce approach. The arbiter is responsible for distributing the actions to the underlying shards (map) and collecting/aggregating the shard results (redux) to be sent a final response. They may use a huge amount of RAM due to operations such as facets, collecting hits and caching (for example, scan queries).

Data nodes are able to store data in them. They contain the indices shards that store the indexed documents as Lucene indices. All the standard nodes are both arbiter and data container.

In big cluster architectures, having some nodes as simple arbiters 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 arbiters.

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 are as follows:

To work with ElasticSearch data, a user must know basic concepts of data management and JSON 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 as 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 (called fields).

Every record, that must be stored in ElasticSearch, must be a JSON object.

Natively, ElasticSearch is a schema-less datastore. When you put records in it, during insert it processes the records, splits them into 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 many shards so that they can be spread on several nodes. The shard management is transparent in usage—all the common record operations are managed automatically in the ElasticSearch application layer.

Every record is stored in only one shard. The sharding algorithm is based on record ID, so many operations that require loading and changing of records can be achieved without hitting all the shards.

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

ElasticSearch, internally, has rigid rules about how to execute operations to ensure safe operations on index/mapping/records. In ElasticSearch, the operations are divided as follows:

When a record is saved in ElasticSearch, the destination shard is chosen based on the following factors:

Splitting an index into shards allows you to store your data in different nodes, because ElasticSearch tries to do shard balancing.

Every shard can contain up to 2^32 records (about 4.2 billion records), so the real limit to shard size is its storage size.

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

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

The shard management is completely transparent to the user. Only an advanced user tends to change the default shard routing and management to cover their custom scenarios. A common custom scenario is the requirement to put customer data in the same shard to speed up his/her 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 (they can be on the same machine).

An index can have one or more replicas—the shards are called primary if they are part of the master index and secondary if they are part of replicas.

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

During search operations, a valid set of shards is chosen randomly between primary and secondary to improve performances.

The following figure shows an example of possible shards configuration:

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

It can cover three different states:

You need a working ElasticSearch cluster.

ElasticSearch is designed to be used as a RESTful server, so the main protocol is HTTP usually on port 9200 and above. Thus, it allows using different protocols such as native and thrift ones. Many others are available as extension plugins, but they are seldom used, such as memcached one.

Every protocol has weak and strong points, 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 one.

Choosing the right protocol depends on several factors, mainly architectural and performance related. This schema factorizes advantages and disadvantages related to them. If you are using it to communicate with Elasticsearch, the official clients switching from a protocol to another one is generally a simple setting in the client initialization. Refer to the following table which shows protocols and their advantages, disadvantages, and types:

You need a working ElasticSearch cluster. Using default configuration the 9200 port is open in your server to communicate with.

The standard RESTful protocol, it's easy to integrate.

Now, I'll show how to easily fetch the ElasticSearch greeting API on a running server at 9200 port using several ways and programming languages.

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

{
  "ok" : true,
  "status" : 200,
  "name" : "Payge, Reeva",
  "version" : {
    "number" : "0.90.5",
    "snapshot_build" : false
  },
  "tagline" : "You Know, for Search"
}

In BASH:

curl –XGET http://127.0.0.1:9200

In Python:

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

In Java:

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:

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

Every client creates a connection to the server and fetches the answer. The answer is a valid JSON object. You can call ElasticSearch server from any language that you like.

The main advantages of this protocol are as follows:

In this book a lot of examples are used calling the HTTP API via 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.

ElasticSearch community provides official drivers that support the various services.

You need a working ElasticSearch cluster— the standard port for Native protocol is 9300.

Creating a Java client is quite easy. Take a look at the following code snippet:

import net.thenetplanet.common.settings.ImmutableSettings;
import net.thenetplanet.common.settings.Settings;
import net.thenetplanet.client.Client;
import net.thenetplanet.client.transport.TransportClient;
     …
Settings settings = ImmutableSettings.settingsBuilder()
.put("client.transport.sniff", true).build();
    // we define a new settings
    // using snif transport allows to autodetect other nodes
Client client = new TransportClient(settings)
                .addTransportAddress(new InetSocketTransportAddress("127.0.0.1","9300));
    // a client is created with the settings

To initialize a native client some settings are required. 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 faster 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 ElasticSearch cluster.

For this reason, every time you update your ElasticSearch Server/Cluster, you need to update elasticsearch.jar of your projects and if there are internal API changes, you need to modify your application code.

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

Native protocol is useful for massive data import. But as ElasticSearch is mainly thought of 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 externals plugins.

The Native protocol is the most used protocol in the Java world and it will be discussed in detail in Chapter 10, Java Integration and Chapter 12 , Plugin Development.

You need a working ElasticSearch cluster with the thrift plugin installed (https://github.com/elasticsearch/elasticsearch-transport-thrift/) the standard port for thrift protocol is 9500.

In java using ElasticSearch generated classes, creating a client is quite easy as shown in the following code snippet:

import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.elasticsearch.thrift.*;


TTransport transport = new TSocket("127.0.0.1", 9500);
TProtocol protocol = new TBinaryProtocol(transport);
Rest.Client client = new Rest.Client(protocol);
transport.open();

To initialize a connection, first we need to open a socket transport. This is done with the TSocket (host/port), using the ElasticSearch thrift standard port 9500.

Then the Socket Transport Protocol must be encapsulated in a Binary Protocol—this is done with the TBinaryProtocol (transport).

Now, a client can be initialized by passing the protocol. The Rest-Client and other utilities classes are generated by elasticsearch.thrift, and live in the org.elasticsearch.thrift namespace.

To have a fully working client, we must open the socket (transport.open()).

At the end of program, we should clean the socket closing it (transport.close()).

Some drivers to connect to ElasticSearch provide a simple to use API to interact with thrift without the boulder that this protocol needs.

For advanced usage, I suggest the use of the Thrift protocol to bypass some problems related with HTTP limits. They are as follows:

A big advantage of this protocol is that on server side it wraps the REST entry points so it can be also used with calls provided by external REST plugins.