ReIndexing Elasticsearch in Scala

The following scala script reads from one index and writes to another script using Scan and scroll method. The script also takes in a partial function where the values from one index can be manipulated before saving into another index. This script assumes you have a field called "id" and an field called "submitDate" so it can continually perform scan and scroll once the preliminary index copy is done, so keep the index's in sync. Notes:

• The ESClient is an extension of on wabisabi Library for elasticsearch
• The Actor initially performs a scan-scroll with submit date gte 1900
• Once the initial scan-scroll is done, it pauses for a minute and performs a scan-scroll again with the submitDate of previous endTime (dateTime.now minus 1 minute)
• This way every minute after the previous run it will continually keep the index in sync
• The partial function "processData" provides a way to manipulate the original data, manipulate it and save it to the new index
• Bulk-indexing is used for saving to the new index, hence a the "id" field is required to determine the "id" of the new document

Usage:

Elasticsearch - Cautionary and Useful Tips

Update/Delete Gotcha:

In elasticsearch, an update to a document is basically a delete and reinsert. A delete operation in elasticsearch is basically marking the document to be deleted and not actually deleted. This is problem especially when you have heavy updates/delete operations as the documents are not actually purged but instead just marked for deletion, which takes up disk space. Following screen shot is an example where the total number of documents in the index (where documents can be searched) is not the same as the actual total documents in the index.



To reclaim disk space, you have to optimize the index: More information at: http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-optimize.html

Memory Limitation - Max ES Heap size:

By default elasticsearch allocates 1GB of heap to it's process. This is ok for development purposes but in production you should generally provide half of the server memory to Elasticsearch. To set the heap size: More the memory given to elasticsearch is better as more data is held in the memory for faster search/seeks but there are few gotchas to be aware of:

• Do not cross 32GB
  As it turns out, the JVM uses a trick to compress object pointers when heaps are less than ~32 GB. Once you cross that magical ~30–32 GB boundary, the pointers switch back to ordinary object pointers. The size of each pointer grows, more CPU-memory bandwidth is used, and you effectively lose memory.
• Give half of the memory to Lucene
  Lucene is designed to leverage the underlying OS for caching in-memory data structures. If you give all available memory to Elasticsearch’s heap, there won’t be any left over for Lucene. This can seriously impact the performance of full-text search.
• Disable Memory Swapping
  Swapping main memory to disk will basically affect server and elasticsearch performance. If memory swaps to disk, a 100-microsecond operation becomes one that take 10 milliseconds. To avoid this you should enable mlockall. This allows the JVM to lock its memory and prevent it from being swapped by the OS. In your elasticsearch.yml, set this:

More information at : http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/heap-sizing.html

Index name - Alias

It's always advisable to provide an alias to the index and have the application use the alias name instead of the actual index name. This is useful as we can switch index's without affecting the calling application. For example, we can create a brand new index with new mappings and basically delete the alias from the old index and assign it to the new index. This way re-indexing operations would be a zero downtime operations.

Logging - Debug/Info/Error:

By default, elasticsearch log level is set to debug in the logging.yml file. This is probably not a good choice as ES tends to log everything which takes up a lot of disk space. I learned this the hard way where I copied data from one index to another index for reindexing purposes and elasticsearch logged every single payload and the log size was almost the size of the index itself! It;s best to have the log level to WARN instead of DEBUG or INFO.

Document Versioning:

For every insert or document update, elasticsearch either auto assigns a version number of expects the user to provide a version number. This is useful for concurrency control. There are 4 types of version mechanism:

• internal: (Default) Auto assigned by elasticsearch
• external: Version number provided by user. Must always be greater than the existing version of the document
• external_gte: Version number provided by user but the version number should be at the very least equal to the existing document version.
• force: Version number provided by the user where the number can be anything. But not recommended

The above four version options are not very well documented, but can be understood by reading the ES source code at: Source

Elasticsearch - No downtime reindexing

As you probably know that mappings in elasticsearch cannot be changed, for example like changing a property type from a string to an int etc. The only way to make such changes is to copy the entire index into a brand new index with new mappings.

Reindexing is an unavoidable common practice as data model changes effects how data is indexed in elastic search. So while designing the system, having an alias assigned to all indexes is a good choice as we can swap indexes in and out. Alias is basically providing an alternate name to an index. For example:



Now all that you need to do is to create a new index with new mappings and copy the data over from the original index to the new index. To perform a bulk copy operation, I prefer to use tools such as elasticsearch-dump which helps in this bulk copy operation.

Following query performs a copy from one keyword index to a second index: Now all that you need to do is to delete the alias from the original index and assign it the new index. This way the calling client using the alias for querying and indexing would have no impact But what about the documents that were updated during the scan and scroll process? Well, that's tricky but if your model does have a update date property you can always re run the es dump to fetch only the documents that were updated after a certain date time.

Elasticsearch - Dynamic Data Mapping

Data in Elasticsearch can be indexed without providing any information about it's content as ES accepts dynamic properties and ES detects if the property value is a string, integer, datetime, boolean etc. In this article, lets work on getting dynamic mapping setup the right way along with some commonly performed search operations.

To start with a simple example. Lets consider the following object: Given the above Json blob and indexing into the elasticsearch would result in the following mapping: This is great that Elasticsearch automatically detected id to be long and text & type to be a string. But if you look carefully, the keywordText and KeywordType are set to default type of "analyzed". This means that those two fields are now available for partial text search. But I want keywordType to be "not_analyzed" as users would never partial text search it. To overcome this but preserve the dynamic nature of this index, we can create a Keywords Index with mapping provided for certain fields: As you can see from above, we have set dynamic to "true" but let the index know that if any field that matches "keywordType" to use a specific mapping rather instead of ES figuring it out for us.

Now that we have KeywordType to "not_analyzed" which basically now is an "exact match" search including the case (upper and lower case). But how do I make KeywordType to be a case insensitive exact match? One way is to to lower the keywordType and have the calling system provide a lower case searches only. For this, the following mapping changes need to happen: We are basically using the "Keyword tokenizer" that Elasticsearch provides that makes it exact match search and "filter" of lowercase which automatically converts the input to lower case. More info on at Elasticsearch tokensizers

So far so good but I dont want users to search on all fields which Elasticsearch by default provides. I would rather have the user provide which field they want to search on. Why? Doing a "_all" search on an index with 100's of fields is a very expensive operation; that's why! To disable "_all" search: OK great, now that _all field search is disabled but now since dynamic is turned on which means any new fields can automagically be indexed, I don't want elasticsearch to index any binary blob as it would consume too much memory; but rather just store it and not index it. For this, the updated mapping would look like: Setting "enabled: false" lets elasticsearch know that this field should not be indexed for search purposes but would be part of the document result. So basically it's stored but not searchable.

Since dynamic mapping is enabled, Elasticsearch parses thru every single property to determine it;s type. As much I would love for Elasticsearch to perform all the magic mappings, let give it a helping hand by letting the Elasticsearch know that certain properties are DateTime type based on it;s name. So basically, if any property that has either "date" or "Date" at it's ending then assume it's a DateTime object. For example "createDate" or "updateDate" would match the above template. Also as you may notice, "date_detection" is set to false.

How about make all string into exact lower case match? So providing dynamic templates when the properties are unknown helps a lot and not have every single field "analyzed" which takes up too memory and extra processing time. The memory consumption analysis will be for another blog post.

Just as an extra, Elasticsearch provides a way to match templates to index names. This means that we provide elasticsearch a template file with mapping information and when an index is created, ES automatically matches the index name with the give template and auto applies the mappings. This template needs to be saved in the "/etc/elasticsearch/templates" folder. An example template file:

Elasticsearch - Zen, AWS Cluster Setup

In a cluster environment, multiple elasticsearch nodes/servers join to form a cluster where the shards are distributed and replicated among these servers but to the outside world it is presented as a single system. For elasticsearch to connect to different nodes, ES provides two discovery methods. One being Zen discovery and other being cloud based discovery via plugins for Azure, AWS and Google Compute engine.

Zen Discovery

From the above snippet, it's pretty straightforward to understand that the discover.type is "zen" and the minimum number of nodes required to form a cluster is "2" and using "unicast" to find other hosts and provide some sort of recovery mechanism (fault detection) if a server went offline or some network problems. This is probably all that Zen discovery has to offer, simple and easy! More info at Zen discovery

AWS/EC2 Discovery

For EC2 discovery, we first need to install the cloud-aws plugin if not already installed From the above config, the discovery type is ec2 and optionally given a region for the plugin to discover other nodes and security group. If there is no IAM role associated with the server, then AWS secret_key and access_key needs to be provided in-order for the plugin to query AWS for node information.

Having the node.auto_attributes set to true would add aws_availability_zone to the node attributes properties which helps in node awareness. What this means is that, given an index with replication factor of 1, ES uses this attribute to determine which node this particular shard is sitting on but makes sure the replicated shard is on a different box. More info at Shard Awareness We can make the Elasticsearch node discovery a little bit faster by filtering the number of servers it needs to ping during the discovery process. This filter can achieved by using the ec2.tag if they are assigned to EC2 servers. In a enterprise environment where there are 100's of ec2 servers deployed on AWS, pinging every single one of them would take a very long time, this should help speed things up. More information for this EC2 discovery plugin at cloud-aws and various discovery at elasticsearch-dsicovery

Elasticsearch - Advanced settings and Tweaks

Now that we have Elasticsearch installed and confirmed working, we can start looking into more advanced settings, more of tweaking, to improve Elasticsearch performance. For most use cases, following three area's of Elasticsearch configuration needs to be addressed:

• Memory configuration
• Threadpool configuration
• Data Store configuration

Memory configuration:

By default Elasticsearch assigns the minimum heap size of 256MB and 1GB maximum heap size. But in real world server environments with many gb in memory availablity, it;s always good to provide 50% of the server memory as a rule of thumb to Elasticsearch process. This setting can be set using: But providing the heap size is just not enough as the memory can be swapped out by the OS. To prevent this we need to lock the process address space assigned to Elasticsearch. This can be done by adding the following line to elasticsearch.yml file and restarting elasticsearch: After starting Elasticsearch, you can see whether this setting was applied successfully by checking the value of mlockall in the output from this request: But the mlockall is false. If you see that mlockall is false, then it means that the the mlockall request has failed. The most probable reason is that the user running Elasticsearch doesn’t have permission to lock memory. This can be granted by running ulimit -l unlimited as root before starting Elasticsearch.
Note that you will always have to run ulimit -l unlimited before elasticsearch restart or else mlockall is set back to false, this is probably because the the User ESprocess is running on is not root

Threadpool Configuration:

Elasticsearch can holds several thread pools with a queue bound to each of these pools which allow pending requests to be held instead of discarded. For example, by default for index operation, it has a fixed thread pool size of # no of processors in the system and a queue_size of 200. So if there are more than 200 requests, the new requests are discarded and following exception is returned back to the client: EsRejectedExecutionException[rejected execution (queue capacity 200)..]
To overcome this limitation and increase the concurrency of elasticsearch processing messages, following setting are be tweaked: So if the use cases if primarily for searching i.e. more search operations than indexing operations, the threadpool for search can be increased and the threadpool for indexing can be much lower. Though queuing up thousands of messages is probably not a wise decision, so tweak responsibly. More information about threadpool size and configuration can be found at Elasticsearch Threadpool

ES by default assumes that you're going to use it mostly for searching and querying, so it allocates 90% of its allocated total HEAP memory for searching. This can be changed with the following settings. Note that implication of this setting can be significant as you are reducing the memory allocated for search purposes! More at Indices Module

Store and indices Configuration:

The store module allows you to control how index data is stored. The index can either be stored in-memory (no persistence) or on-disk (the default). Unless your data is temporary data using in-memory store is a bad idea as you will loose the data upon restart. For disk based storage, we need to have fast disk seeks if the data to be looked up is not in memory. The most optimal way is to use mmap fs which is basically memory mapped files. More information regarding storage options can be found at Elasticsearch Store

Elasticsearch - Installation and general settings

Installation of Elasticsearch is a breeze by which I mean it's as simple as downloading the zip/tar file and unzipping it.
In the above bash script, we are essentially downloading the file, unzipping and installing couple of plugins for administration and cloud discovery. Now that we have Elasticsearch unzipped, we can optionally provide location to it's data, log and configuration folder. There are two ways to provide Elastisearch this configuration. First way is to provide the paths in the yml configuration file elasticsearch.yml. For example Second way is running Elasticsearch in daemon mode, you can setup the paths in the sysconfig file generally located at /etc/sysconfig/elasticsearch. The configuration in this file is passed onto ES as command line settings when elasticsearch is started. Note that the configuration in the elasticsearch.yml file overrides the sysconfig file
But lets say that we using a package manager or puppet scripts to install elasticsearch and now we have no idea where the config files and data directories located. One easy way to get these information is to curl the elasticsearch node endpoint which returns back all the information regarding each node with all path and configuration information More information on the directory structure can be found at Elasticsearch Directory layout
OK, now that we have elasticsearch unzipped and the data directory setup, lets update some minimal but essential Elasticsearch configuration: List of all configuration can be found at: Elasticsearch configuration file
Note that if the node.name is not provided, Elasticsearch automatically assigns a node name based on Marvel comic characters. This is fine as long as Elasticsearch process does not restart as it will assign a new name again which could be trouble if you are monitoring the ES process by node names.
Now that we have the basic elasticsearch settings updated/added, we can start elasticsearch by running:

Log analysis using Logstash, ElasticSearch and Kibana

Introduction:

	Logstash is a free tool for managing events and logs. It has three primary components, an Input module for collecting logs from various sources [http://logstash.net/docs/1.1.13/], a parsing module for tweaking and parsing data and finally a storage/output module to save or pass along the parsed data to other systems [http://logstash.net/docs/1.1.13/].
	ElasticSearch is this awesome distributable, RESTful, free Lucene powered search engine/server. Unlike SOLR, ES is very simple to use and maintain and similar to SOLR, indexing is near realtime.
	Kibana is a presentation layer that sits on top of Elasticsearch to analyze and make sense of logs that logstash throws into Elastic search; Kibana is a highly scalable interface for Logstash and ElasticSearch that allows you to efficiently search, graph, analyze and otherwise make sense of a mountain of logs.

Logstash + ElasticSearch + Kibana combination can be compared to open sourced Splunk but on a smaller scale.

Setup:
Logstash, is as easy as downloading [http://logstash.net/] the JAR file and setting up the input and ouput sources and running the java command. In this example, I will be monitoring a log file and writing it into Elasticsearch server for users to analyse the data using Kibana.


Elasticsearch: download the zip package from the site [http://www.elasticsearch.org/download/] and run the elasticsearch.bat file.
Note: make sure the JAVA_HOME is setup up right for the logstash and elasticsearch to work.



Kibana: download the kibana files from Github [https://github.com/elasticsearch/kibana] and either run it as a standalone app or make it part of ElasticSearch plugins. You can do this by copying the kibana files to the ElasticSearch plugins / sites directory.



*Open config.js in your favorite editor
*Set elasticsearch: 'http://localhost:9200', to your ElasticSearch server

Use case:
In general most of these log analyzer always talk about analyzing website traffic etc similar to the videos that Kibana has on their website. [http://kibana.org/about.html]. This is great but in real world logs and events are more than just website traffic such as information flow checkpoints, performance data etc.
In our case, lets assume we have some data that is being passed from one system to another and we are logging to a file. A simple representation of this information flow is as follows:



So basically there are 4 systems or states that the data is passed thru, Ingest, Digest, Process and Exit. At each of these systems, an event is logged to track the data flow or basically checkpoints. These events are logged in dataLog.log file as mentioned in the above logstash configuration file.

Once the logstash is up and running, logstash basically tails the files and copies the logged events to elastic search as JSON objects. Elasticsearch index;s all the fields and kibana is now ready to access the data. Following are some of the cases that can be analyzed using Kibana:

Show all data flowing thru the system	Filter by Id
Get All Error'd	Advanced Filter using Lucene Syntax

The above reporting/analysis are just a few examples that can be achieved using Kibana + Elasticsearch. With Kibana you can design your own custom dashboards with configurable panels that can be grouped by role. Charts and panels are fully interactive with features like drill down, range selection and customization. With using Elasticsearch, rapid data growth is as easy as adding more ES servers (in cluster).