Elasticsearch for Hadoop

The examples in this book are developed and tested against Oracle Java 1.8. These examples should run fine with other distributions of Java 8 as well.

In order to set up Oracle Java 8, open the terminal and execute the following steps:

To ensure that our ES-Hadoop environment is clean and isolated from the rest of the applications and to be able to manage security and permissions easily, we will set up a dedicated user. Perform the following steps:

Now, you need to relogin with the eshadoop user to execute further steps.

In order to manage nodes, Hadoop requires an SSH access, so let's install and run the SSH. Perform the following steps:

Using the following commands, download Hadoop and extract the file to /usr/local so that it is available for other users as well. Perform the following steps:

The next step is to set up environment variables. You can do so by exporting the required variables to the .bashrc file for the user.

Open the .bashrc file using any editor of your choice, then add the following export declarations to set up our environment variables:

#Set JAVA_HOME 
export JAVA_HOME=/usr/lib/jvm/java-8-oracle

#Set Hadoop related environment variable
export HADOOP_INSTALL=/usr/local/hadoop

#Add bin and sbin directory to PATH
export PATH=$PATH:$HADOOP_INSTALL/bin
export PATH=$PATH:$HADOOP_INSTALL/sbin

#Set few more Hadoop related environment variable
export HADOOP_MAPRED_HOME=$HADOOP_INSTALL
export HADOOP_COMMON_HOME=$HADOOP_INSTALL
export HADOOP_HDFS_HOME=$HADOOP_INSTALL
export YARN_HOME=$HADOOP_INSTALL
export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_INSTALL/lib/native
export HADOOP_OPTS="-Djava.library.path=$HADOOP_INSTALL/lib"

Once you have saved the .bashrc file, you can relogin to have your new environment variables visible, or you can source the .bashrc file using the following command:

$ source ~/.bashrc

Now, we need to set up the JAVA_HOME environment variable in the hadoop-env.sh file that is used by Hadoop. You can find it in $HADOOP_INSTALL/etc/hadoop.

Next, change the JAVA_HOME path to reflect to your Java installation directory. On my machine, it looks similar to the following:

$ export JAVA_HOME=/usr/lib/jvm/java-8-oracle

Now, let's relogin and confirm the configuration using the following command:

$ hadoop version

As you know, we will set up our Hadoop environment in a pseudo-distributed mode. In this mode, each Hadoop daemon runs in a separate Java process. The next step is to configure these daemons. So, let's switch to the following folder that contains all the Hadoop configuration files:

$ cd $HADOOP_INSTALL/etc/hadoop

<configuration>
<property>
  <name>hadoop.tmp.dir</name>
  <value>/home/eshadoop/hdfs/tmp</value>
  <description>A base for other temporary directories.</description>
 </property>
<property>
   <name>fs.default.name</name>
   <value>hdfs://localhost:9000</value>
</property>
</configuration>

<configuration>
   <property>
   <name>dfs.replication</name>
   <value>1</value>
 </property>
</configuration>

<configuration>
<property>
   <name>yarn.nodemanager.aux-services</name>
   <value>mapreduce_shuffle</value>
</property>
<property>
   <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>
   <value>org.apache.hadoop.mapred.ShuffleHandler</value>
</property>
</configuration>

<configuration>  
  <property>
    <name>mapred.job.tracker</name>
    <value>yarn</value>
  </property>
</configuration>

$ hadoop namenode -format

Now, we have all the prerequisites set up along with all the Hadoop daemons. In order to run our first MapReduce job, we need all the required Hadoop daemons running.

Let's start with HDFS using the following command. This command starts the NameNode, SecondaryNameNode, and DataNode daemons:

$ start-dfs.sh

The next step is to start the YARN resource manager using the following command (YARN will start the ResourceManager and NodeManager daemons):

$ start-yarn.sh

If the preceding two commands were successful in starting HDFS and YARN, you should be able to check the running daemons using the jps tool (this tool lists the running JVM process on your machine):

$ jps

If everything worked successfully, you should see the following services running:

13386 SecondaryNameNode
13059 NameNode
13179 DataNode
17490 Jps
13649 NodeManager
13528 ResourceManager

To download Elasticsearch, perform the following steps:

The Elasticsearch configuration file, elasticsearch.yml, can be located in the config folder under the Elasticsearch home directory. Open the elasticsearch.yml file in the editor of your choice by using the following command:

$ cd /usr/local/elasticsearch
$ vi config/elasticsearch.yml

Uncomment the line with the cluster.name key from the elasticsearch.yml file and change the cluster name, as shown in the following code:

  cluster.name:eshadoopcluster

Similarly, uncomment the line with the node.name key and change the value as follows:

node.name:"ES Hadoop Node"

You can change the defaults for configurations starting with path.*. To set up the directories that store the server data, to locate paths section, and to uncomment the highlighted paths and changes, use the following code:

 ########################### paths #############################
 # Path to directory containing configuration (this file and logging.yml):
#
path.conf: /usr/local/elasticsearch/config

# Path to directory where to store index data allocated for this node.
# 
# Can optionally include more than one location, causing data to be striped across
# the locations (a la RAID 0) on a file level, favouring locations with most free
# space on creation. 
path.data: /usr/local/elasticsearch/data


# Path to temporary files:
#
path.work: /usr/local/elasticsearch/work

# Path to log files:
#
path.logs: /usr/local/elasticsearch/logs

Elasticsearch provides a plugin utility to install the Elasticsearch plugins. Execute the following command to install the Head plugin:

$ bin/plugin -install mobz/elasticsearch-head

-> Installing mobz/elasticsearch-head...
Trying https://github.com/mobz/elasticsearch-head/archive/master.zip...
Downloading ..................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................DONE
Installed mobz/elasticsearch-head into /usr/local /elasticsearch/plugins/head
Identified as a _site plugin, moving to _site structure ...

As indicated by the console output, the plugin is successfully installed in the default plugins directory under the Elasticsearch home. You can access the head plugin at http://localhost:9200/_plugin/head/.

Now, let's install the Marvel plugin using a similar command:

$ bin/plugin -i elasticsearch/marvel/latest

-> Installing elasticsearch/marvel/latest...
Trying http://download.elasticsearch.org/elasticsearch/marvel/marvel-latest.zip...
Downloading ................................................................................................................................................................................................................................................................................................................................DONE
Installed elasticsearch/marvel/latest into /usr/local/elasticsearch/plugins/marvel

Finally, start Elasticsearch using the following command:

$ ./bin/elasticsearch

We will then get the following log:

[2015-05-13 21:59:37,344][INFO ][node                     ] [ES Hadoop Node] version[1.5.1], pid[3822], build[5e38401/2015-04-09T13:41:35Z]
[2015-05-13 21:59:37,346][INFO ][node                     ] [ES Hadoop Node] initializing ...
[2015-05-13 21:59:37,358][INFO ][plugins                  ] [ES Hadoop Node] loaded [marvel], sites [marvel, head]
[2015-05-13 21:59:39,956][INFO ][node                     ] [ES Hadoop Node] initialized
[2015-05-13 21:59:39,959][INFO ][node                     ] [ES Hadoop Node] starting ...
[2015-05-13 21:59:40,133][INFO ][transport                ] [ES Hadoop Node] bound_address {inet[/0:0:0:0:0:0:0:0:9300]}, publish_address {inet[/10.0.2.15:9300]}
[2015-05-13 21:59:40,159][INFO ][discovery                ] [ES Hadoop Node] eshadoopcluster/_bzqXWbLSXKXWpafHaLyRA
[2015-05-13 21:59:43,941][INFO ][cluster.service          ] [ES Hadoop Node] new_master [ES Hadoop Node][_bzqXWbLSXKXWpafHaLyRA][eshadoop][inet[/10.0.2.15:9300]], reason: zen-disco-join (elected_as_master)
[2015-05-13 21:59:43,989][INFO ][http                     ] [ES Hadoop Node] bound_address {inet[/0:0:0:0:0:0:0:0:9200]}, publish_address {inet[/10.0.2.15:9200]}
[2015-05-13 21:59:43,989][INFO ][node                     ] [ES Hadoop Node] started
[2015-05-13 21:59:44,026][INFO ][gateway                  ] [ES Hadoop Node] recovered [0] indices into cluster_state
[2015-05-13 22:00:00,707][INFO ][cluster.metadata         ] [ES Hadoop Node] [.marvel-2015.05.13] creating index, cause [auto(bulk api)], templates [marvel], shards [1]/[1], mappings [indices_stats, cluster_stats, node_stats, shard_event, node_event, index_event, index_stats, _default_, cluster_state, cluster_event, routing_event]
[2015-05-13 22:00:01,421][INFO ][cluster.metadata         ] [ES Hadoop Node] [.marvel-2015.05.13] update_mapping [node_stats] (dynamic)

The startup logs will give you some useful hints as to what is going on. By default, Elasticsearch uses the transport ports from 9200 to 9299 for HTTP, allocating the first port that is available for the node. In the highlighted output, you can also see that it binds to the port 9300 as well. Elasticsearch uses the port range from 9300 to 9399 for an internal node-to-node communication or when communicating using the Java client. It can use the zen multicast or the unicast ping discovery to find other nodes in the cluster with multicast as the default. We will understand more about these discovery nodes in later chapters.

You can download the examples in the book from https://github.com/vishalbrevitaz/eshadoop/tree/master/ch01. Once you have got the source code, you can build the JAR file for this chapter using the steps mentioned in the readme file in the source code zip. The build process should generate a ch01-0.0.1-job.jar file under the <SOURCE_CODE_BASE_DIR>/ch01/target directory.

For our WordCount example, you can use any text file of your choice. To explain the example, we will use the sample.txt file that is part of the source zip. Perform the following steps:

We are ready with the job JAR file; its sample file is imported to HDFS. Point your terminal to the <SOURCE_CODE_BASE_DIR>/ch01/target directory and run the following command:

$ hadoop jar ch01-0.0.1-job.jar /input/ch01/sample.txt  

Now you'll get the following output:

15/05/10 15:21:33 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/05/10 15:21:34 WARN mr.EsOutputFormat: Speculative execution enabled for reducer - consider disabling it to prevent data corruption
15/05/10 15:21:34 INFO util.Version: Elasticsearch Hadoop v2.0.2 [ca81ff6732]
15/05/10 15:21:34 INFO mr.EsOutputFormat: Writing to [eshadoop/wordcount]
15/05/10 15:21:35 WARN mapreduce.JobSubmitter: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
15/05/10 15:21:41 INFO input.FileInputFormat: Total input paths to process : 1
15/05/10 15:21:42 INFO mapreduce.JobSubmitter: number of splits:1
15/05/10 15:21:42 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1431251282365_0002
15/05/10 15:21:42 INFO impl.YarnClientImpl: Submitted application application_1431251282365_0002
15/05/10 15:21:42 INFO mapreduce.Job: The url to track the job: http://eshadoop:8088/proxy/application_1431251282365_0002/
15/05/10 15:21:42 INFO mapreduce.Job: Running job: job_1431251282365_0002
15/05/10 15:21:54 INFO mapreduce.Job: Job job_1431251282365_0002 running in uber mode : false
15/05/10 15:21:54 INFO mapreduce.Job:  map 0% reduce 0%
15/05/10 15:22:01 INFO mapreduce.Job:  map 100% reduce 0%
15/05/10 15:22:09 INFO mapreduce.Job:  map 100% reduce 100%
15/05/10 15:22:10 INFO mapreduce.Job: Job job_1431251282365_0002 completed successfully
…
…
…
  Elasticsearch Hadoop Counters
    Bulk Retries=0
    Bulk Retries Total Time(ms)=0
    Bulk Total=1
    Bulk Total Time(ms)=48
    Bytes Accepted=9655
    Bytes Received=4000
    Bytes Retried=0
    Bytes Sent=9655
    Documents Accepted=232
    Documents Received=0
    Documents Retried=0
    Documents Sent=232
    Network Retries=0
    Network Total Time(ms)=84
    Node Retries=0
    Scroll Total=0
    Scroll Total Time(ms)=0

We just executed our first Hadoop MapReduce job that uses and imports data to Elasticsearch. This MapReduce job simply outputs the count of each word in the Mapper phase, and Reducer calculates the sum of all the counts for each word. We will dig into greater details of how exactly this WordCount program is developed in the next chapter. The console output of the job displays the useful log information to indicate the progress of the job execution. It also displays the ES-Hadoop counters that provide some handy information about the amount of data and documents being sent and received, the number of retries, the time taken, and so on. If you have used the sample.txt file provided in the source zip, you will be able to see that the job found 232 unique words and all of them are pushed as the Elasticsearch document. In the next section, we will examine these documents with the Elasticsearch Head and Marvel plugin that we already installed in Elasticsearch. Note that you can also track the status of your ES-Hadoop MapReduce jobs, similar to any other Hadoop jobs, in the job tracker. In our setup, you can access the job tracker at http://localhost:8088/cluster.

The Elasticsearch Head plugin provides a simple web frontend to visualize the Elasticsearch indices, cluster, node health, and statistics. It provides an easy-to-use interface to explore index, types, and documents with the query building interface. It also allows you to view the documents of Elasticsearch in a table-like structure as well, which can be quite handy for users coming from the RDBMS background.

Here is how the Elasticsearch Head home page looks when you open http://localhost:9200/_plugin/head.

The following image shows the home page of the Elasticsearch Head plugin:

You will get a quick insight into your cluster from the preceding screenshot, such as what is cluster health is (Green, Yellow, or Red); how the shards are allocated to different nodes, which indices exist in the cluster, what the size is of each index, and so on. For example, in the preceding screenshot, we can see two indices: .marvel-2015.05.10 and eshadoop. You may be surprised that we never created an index with the name of .marvel-2015.05.10. You can ignore this index for the time being; we will take a brief look at it in the next subsection.

Let's go back to our WordCount example. You can see that the document count for the eshadoop index in the preceding screenshot exactly matches with the number of documents metric indicated by the MapReduce job output that we saw in the last section.

The following diagram shows the Browser tab of the Elasticsearch Head plugin:

To take a look at the documents, navigate to the Browser tab. You can see that the screen is similar to the one shown in the preceding screenshot. You can click on the eshadoop index on the left-hand side under the Indices heading and sort the results by count to see the relevant documents. You can also see that the output of the MapReduce job is pushed directly to Elasticsearch. Further more, you can see the ES document fields, such as _index, _type, _id, and _score, along with the fields that we are interested in word and count. You may want to sort the results based on count by clicking on the count column to see the most frequent words in the sample.txt file.

Marvel is a monitoring dashboard for real-time and historical analysis that is built on top of Kibana: a data visualization tool for ES-Hadoop. This dashboard provides, insight into the different metrics of the node, JVM, and ES-Hadoop internals. To open the Marvel dashboard, refer to your browser at http://localhost:9200/_plugin/marvel/.

The following screenshot gives you an overview of the Marvel dashboard:

You can see the different real-time metrics for your cluster, nodes, and indices. You can visualize the trends of the document count, search, and the indexing request rates in a graphical way. This kind of visualization may be helpful to get a quick insight into the usage pattern of the index and find out the candidates for the purpose of performance optimization. It displays the vital monitoring stats, such as the CPU usage, the load, the JVM memory usage, the free disk space, and so on. You can also filter by time range in the top-right corner to use the dashboard for historical analysis. Marvel stores these historical data in a separate daily rolling index with a name pattern, such as .marvel-XXX.

Exploring the data in Sense

Sense is a plugin embedded in Marvel to provide a seamless and easy-to-use REST API client for the ES-Hadoop server. It is Elasticsearch-aware and frees you from memorizing the ES-Hadoop query syntaxes by providing autosuggestions. It also helps by indicating the typo or syntax errors.

To open the Sense user interface, open http://localhost:9200/_plugin/marvel/sense/index.html in your browser.

The following screenshot shows the query interface of Sense:

Now, let's find out the documents imported in the eshadoop index by executing the match_all query.

Then, use the following query in the query panel on the left-hand side in the sense interface:

GET eshadoop/_search
{
   "query": {
        "match_all":{}
    }
}

Finally, click on the Send request button to execute the query and obtain the results.

Note

You can point to different Elasticsearch servers if you wish by changing the server field at the top.