Table of Contents
Introduction
This blog continues our previous blog, How We Build Reporting Systems To Ensure Enhanced Customer Satisfaction.
This blog describes the optimisation of Elasticsearch indices to improve search performance, sustain Elasticsearch’s peak performance and optimise the hardware resources.
Indices
Indices are where data is stored in Elasticsearch. One index is made up of one or many shards.
Shards
Shard is an instance of a Lucene index, which you can think of as a self-contained search engine that indexes and handles queries for a subset of the data of the index it belongs to. There are two types of shards:
- Primary shard: Holds the part of data stored in an index.
- Replica shard: Redundant copies of the primary shard that improve query performance, and protect against hardware failures.
Choosing the number of shards
- A common approach is to start with one shard and keep increasing the number until you achieve the highest performance.
- If you have 2 data nodes, you should have 2, 4, or 6 shards. If you have decided to use 3 primary shards, then your indexing process will actually be slower than when using 2 shards, because 1 node will have double the work.
- 20-25 GB is usually a good shard size for search operations.
Elasticsearch best practices used in reporting services
How do we create indices
We have defined an interval to create a new index at a day, week, fortnight, month, quarter, and year level based on the volume of data in the index.
Multiple primary shards when indexing
When you send a bulk request to index a list of documents, they will be divided among all available primary shards.
For example, if you have 4 primary shards and send a request with 80 documents, each shard will index 20 documents in parallel. Once all documents are indexed, the response is sent back to the client.
Multiple primary shards when searching
How does parallelisation work for searching? Each shard contains only a subset of the documents. When you submit a search request, the node that receives the request acts as the coordinating node, which then looks into the cluster state to find all the shards that make up the index. The coordinating node will then dispatch that query to all of those shards.
After that, each shard locally computes a list of results with document IDs and scores. Those partial results are sent back to the coordinating node which merges them into a single result. Then, the second phase begins, where a list of documents gets fetched.
Why do we need to optimise elastic indices?
Quick exhaustion of the shards
In our reporting service, on average, we create 6 indices every day which takes 40 shards, each index having a different primary and replica shard count based on data volume and traffic. For a year, it will be 6*365=2190 indices and 40*365= 14600 shards and our retention period is 2 years, or even more for some of the indices.
As you see, 14600 shards are required in one year. Each shard utilises some resources for mapping, storing cluster states, querying, etc. The higher the number of shards, the greater the resource utilisation, and keeping too many active shards in the Elasticsearch cluster will degrade the overall performance of Elasticsearch.
Search performance
Usually, writing happens once and reading happens numerous times. In fact, most of the data is immutable a few days after it is written and some of them are immutable immediately.
Let’s assume you have a day-level index order and you will have millions of orders daily. So, you need fast ingestion for a day or for a few days. Then, you will have mostly read traffic.
In the above scenario, you can create the index with more primary shards based on the data volume and ingestion traffic. As you have already seen above, having multiple shards when searching will have performance issues, this is where the index optimisation is required once the index will not incur any more write.
How Did We Optimise Elastic Indices?
We have used shrink API and force-merge API of Elasticsearch to optimise indices.
We have implemented an index optimiser as part of our application which will be running on a scheduled basis when the traffic is less. The optimisation process is as follows:
Update index setting: Before you can shrink an index, the index must be read-only. A copy of every shard in the index must reside on the same node and the index must have a green health status. Also, it is recommended to remove the replicas to make the shard allocation easier.
Shrink index: The shrink index API allows you to shrink an existing index into a new index with fewer primary shards. The requested number of primary shards in the target index must be a factor in the number of shards in the source index. For example, an index with 8 primary shards can be shrunk into 4, 2, or 1 primary shard.
You can monitor shrink processes with cluster health API to wait until all primary shards have been allocated by setting the wait_for_status parameter to yellow or green.
- Add an alias: As you can see from the shrink API, the target index name will be different from your source index name. If your application needs the same name then you can delete the existing source index and add the alias with the same name to the target index.
- Merge segments: You can use the force merge API to reduce the number of segments in each shard by merging some of them together. This step also frees up the space used by deleted documents.
In Lucene, a document is not deleted from a segment but just marked as deleted. During a merge, a new segment is created that does not contain those document deletions.
- Add replica: We removed the replica before shrinking the index. Now, you can add one or more replicas based on your requirement.
As you see, we have to execute multiple steps sequentially on indices as part of optimisation and some of the steps will be executed asynchronously, i.e. update setting, merge shards, merge segments, etc. Hence, we would be needing a way to track the steps of indices that will need optimisation. We have used ActiveMQ queue for the same. We have retry enabled at each step that will help to retry in case of failures.
Below table has the comparison, before and after index optimisation:
Note:
Force merge should only be called against an index after you have finished writing to it. If you continue to write to such an index then the automatic merge policy will never consider these segments for future merges until they mostly consist of deleted documents. This can cause very large segments to remain in the index which can result in increased disk usage and worse search performance.
If you have a use case where you need a write operation after force merges then you can allow the write operation. Then, you can run the below force merge API manually once a month to clean the deleted documents from segments since the automatic merge policy will not be considered.
Conclusion
In this post, we shared our experience on how we optimised the indices to increase the search performance and keep the elastic search running at peak performance with fewer resources.
