Commit 936164e2 authored by Dave Johnson's avatar Dave Johnson
Browse files

Refactoring ansible examples

Parameterized group names for LAMP haproxy example to support multiple inventory providers
Factored out specific host patterns e.g. "hosts: jboss"
Removed specific user declarations e.g. "user: root"
Removed specific references to "sudo"
parents 9c85f13e 375481a2
# Deploying Hadoop Clusters using Ansible
## Preface
The playbooks in this example are designed to deploy a Hadoop cluster on a
CentOS 6 or RHEL 6 environment using Ansible. The playbooks can:
1) Deploy a fully functional Hadoop cluster with HA and automatic failover.
2) Deploy a fully functional Hadoop cluster with no HA.
3) Deploy additional nodes to scale the cluster
These playbooks require Ansible 1.2, CentOS 6 or RHEL 6 target machines, and install
the open-source Cloudera Hadoop Distribution (CDH) version 4.
## Hadoop Components
Hadoop is framework that allows processing of large datasets across large
clusters. The two main components that make up a Hadoop cluster are the HDFS
Filesystem and the MapReduce framework. Briefly, the HDFS filesystem is responsible
for storing data across the cluster nodes on its local disks. The MapReduce
jobs are the tasks that would run on these nodes to get a meaningful result
using the data stored on the HDFS filesystem.
Let's have a closer look at each of these components.
![Alt text](images/hdfs.png "HDFS")
The above diagram illustrates an HDFS filesystem. The cluster consists of three
DataNodes which are responsible for storing/replicating data, while the NameNode
is a process which is responsible for storing the metadata for the entire
filesystem. As the example illustrates above, when a client wants to write a
file to the HDFS cluster it first contacts the NameNode and lets it know that
it want to write a file. The NameNode then decides where and how the file
should be saved and notifies the client about its decision.
In the given example "File1" has a size of 128MB and the block size of the HDFS
filesystem is 64 MB. Hence, the NameNode instructs the client to break down the
file into two different blocks and write the first block to DataNode 1 and the
second block to DataNode 2. Upon receiving the notification from the NameNode,
the client contacts DataNode 1 and DataNode 2 directly to write the data.
Once the data is recieved by the DataNodes, they replicate the block across the
other nodes. The number of nodes across which the data would be replicated is
based on the HDFS configuration, the default value being 3. Meanwhile the
NameNode updates its metadata with the entry of the new file "File1" and the
locations where the parts are stored.
## MapReduce
MapReduce is a Java application that utilizes the data stored in the
HDFS filesystem to get some useful and meaningful result. The whole job is
split into two parts: the "Map" job and the "Reduce" Job.
Let's consider an example. In the previous step we had uploaded the "File1"
into the HDFS filesystem and the file was broken down into two different
blocks. Let's assume that the first block had the data "black sheep" in it and
the second block has the data "white sheep" in it. Now let's assume a client
wants to get count of all the words occurring in "File1". In order to get the
count, the first thing the client would have to do is write a "Map" program
then a "Reduce" program.
Here's a psudeo code of how the Map and Reduce jobs might look:
mapper (File1, file-contents):
for each word in file-contents:
emit (word, 1)
reducer (word, values):
sum = 0
for each value in values:
sum = sum + value
emit (word, sum)
The work of the Map job is to go through all the words in the file and emit
a key/value pair. In this case the key is the word itself and value is always
The Reduce job is quite simple: it increments the value of sum by 1, for each
value it gets.
Once the Map and Reduce jobs are ready, the client would instruct the
"JobTracker" (the process resposible for scheduling the jobs on the cluster)
to run the MapReduce job on "File1"
Let's have closer look at the anotomy of a Map job.
![Alt text](images/map.png "Map job")
As the figure above shows, when the client instructs the JobTracker to run a
job on File1, the JobTracker first contacts the NameNode to determine where the
blocks of the File1 are. Then the JobTracker sends the Map job's JAR file down
to the nodes having the blocks, and the TaskTracker process those nodes to run
the application.
In the above example, DataNode 1 and DataNode 2 havw the blocks, so the
TaskTrackers on those nodes run the Map jobs. Once the jobs are completed the
two nodes would have key/value results as below:
MapJob Results:
"Black: 1"
"Sheep: 1"
"White: 1"
"Sheep: 1"
Once the Map phase is completed the JobTracker process initiates the Shuffle
and Reduce process.
Let's have closer look at the Shuffle-Reduce job.
![Alt text](images/reduce.png "Reduce job")
As the figure above demonstrates, the first thing that the JobTracker does is
spawn a Reducer job on the DataNode/Tasktracker nodes for each "key" in the job
result. In this case we have three keys: "black, white, sheep" in our result,
so three Reducers are spawned: one for each key. The Map jobs shuffle the keys
out to the respective Reduce jobs. Then the Reduce job code runs and the sum is
calculated, and the result is written into the HDFS filesystem in a common
directory. In the above example the output directory is specified as
"/home/ben/output" so all the Reducers will write their results into this
directory under different filenames; the file names being "part-00xx", where x
is the Reducer/partition number.
## Hadoop Deployment
![Alt text](images/hadoop.png "Reduce job")
The above diagram depicts a typical Hadoop deployment. The NameNode and
JobTracker usually reside on the same machine, though they can run on seperate
machines. The DataNodes and TaskTrackers run on the same node. The size of the
cluster can be scaled to thousands of nodes with petabytes of storage.
The above deployment model provides redundancy for data as the HDFS filesytem
takes care of the data replication. The only single point of failure are the
NameNode and the JobTracker. If any of these components fail the cluster will
not be usable.
## Making Hadoop HA
To make the Hadoop cluster highly available we would have to add another set of
JobTracker/NameNodes, and make sure that the data updated by the master is also
somehow also updated by the client. In case of failure of the primary node, the
secondary node takes over that role.
The first thing that has to be dealt with is the data held by the NameNode. As
we recall, the NameNode holds all of the metadata about the filesystem, so any
update to the metadata should also be reflected on the secondary NameNode's
metadata copy. The synchronization of the primary and seconary NameNode
metadata is handled by the Quorum Journal Manager.
### Quorum Journal Manager
![Alt text](images/qjm.png "QJM")
As the figure above shows the Quorum Journal manager consists of the journal
manager client and journal manager nodes. The journal manager clients reside
on the same node as the NameNodes, and in case of primary node, collects all the
edits logs happening on the NameNode and sends it out to the Journal nodes. The
journal manager client residing on the secondary namenode regurlary contacts
the journal nodes and updates its local metadata to be consistant with the
master node. In case of primary node failure the secondary NameNode updates
itself to the latest edit logs and takes over as the primary NameNode.
### Zookeeper
Apart from data consistency, a distributed cluster system also needs a
mechanism for centralized coordination. For example, there should be a way for
the secondary node to tell if the primary node is running properly, and if not
it has to take up the role of the primary. Zookeeper provides Hadoop with a
mechanism to coordinate in this way.
![Alt text](images/zookeeper.png "Zookeeper")
As the figure above shows, the Zookeeper services are client/server baseds
service. The server component itself is replicated over a set of machines that
comprise the service. In short, high availability is built into the Zookeeper
For Hadoop, two Zookeeper clients have been built: ZKFC (Zookeeper Failover
Controller), one for the NameNode and one for JobTracker. These clients run on
the same machines as the NameNode/JobTrackers themselves.
When a ZKFC client is started, it establishes a connection with one of the
Zookeeper nodes and obtains a session ID. The client then keeps a health check
on the NameNode/JobTracker and sends heartbeats to the ZooKeeper.
If the ZKFC client detects a failure of the NameNode/JobTracker, it removes
itself from the ZooKeeper active/standby election, and the other ZKFC client
fences the node/service and takes over the primary role.
## Hadoop HA Deployment
![Alt text](images/hadoopha.png "Hadoop_HA")
The above diagram depicts a fully HA Hadoop Cluster with no single point of
failure and automated failover.
## Deploying Hadoop Clusters with Ansible
Setting up a Hadoop cluster without HA itself can be a challenging and
time-consuming task, and with HA, things become even more difficult.
Ansible can automate the whole process of deploying a Hadoop cluster with or
without HA with the same playbook, in a matter of minutes. This can be used for
quick environment rebuild, or in case of disaster or node failures, recovery
time can be greatly reduced with Ansible automation.
Let's have a look to see how this is done.
## Deploying a Hadoop cluster with HA
### Prerequisites
These playbooks have been tested using Ansible v1.2, and CentOS 6.x (64 bit)
Modify group_vars/all to choose the network interface for Hadoop communication.
Optionally you change the Hadoop-specific parameters like ports or directories
by editing group_vars/all file.
Before launching the deployment playbook make sure the inventory file (hosts)
is set up properly. Here's a sample:
zhadoop1 zoo_id=1
zhadoop2 zoo_id=2
zhadoop3 zoo_id=3
Once the inventory is set up, the Hadoop cluster can be setup using the following
ansible-playbook -i hosts site.yml
Once deployed, we can check the cluster sanity in different ways. To check the
status of the HDFS filesystem and a report on all the DataNodes, log in as the
'hdfs' user on any Hadoop master server, and issue the following command to get
the report:
hadoop dfsadmin -report
To check the sanity of HA, first log in as the 'hdfs' user on any Hadoop master
server and get the current active/standby NameNode servers this way:
-bash-4.1$ hdfs haadmin -getServiceState zhadoop1
-bash-4.1$ hdfs haadmin -getServiceState zhadoop2
To get the state of the JobTracker process login as the 'mapred' user on any
Hadoop master server and issue the following command:
-bash-4.1$ hadoop mrhaadmin -getServiceState hadoop1
-bash-4.1$ hadoop mrhaadmin -getServiceState hadoop2
Once you have determined which server is active and which is standby, you can
kill the NameNode/JobTracker process on the server listed as active and issue
the same commands again, and you should see that the standby has been promoted
to the active state. Later, you can restart the killed process and see that node
listed as standby.
### Running a MapReduce Job
To deploy the mapreduce job run the following script from any of the hadoop
master nodes as user 'hdfs'. The job would count the number of occurance of the
word 'hello' in the given inputfile. Eg: su - hdfs -c "/tmp/"
cat > /tmp/inputfile << EOF
hadoop fs -put /tmp/inputfile /inputfile
hadoop jar /usr/lib/hadoop-0.20-mapreduce/hadoop-examples.jar grep /inputfile /outputfile 'hello'
hadoop fs -get /outputfile /tmp/outputfile/
To verify the result, read the file on the server located at
/tmp/outputfile/part-00000, which should give you the count.
## Scale the Cluster
When the Hadoop cluster reaches its maximum capacity, it can be scaled by
adding nodes. This can be easily accomplished by adding the node hostname to
the Ansible inventory under the hadoop_slaves group, and running the following
ansible-playbook -i hosts site.yml --tags=slaves
## Deploy a non-HA Hadoop Cluster
The following diagram illustrates a standalone Hadoop cluster.
To deploy this cluster fill in the inventory file as follows:
Edit the group_vars/all file to disable HA:
ha_enabled: False
And run the following command:
ansible-playbook -i hosts site.yml
The validity of the cluster can be checked by running the same MapReduce job
that has documented above for an HA Hadoop cluster.
# Defaults to the first ethernet interface. Change this to:
# iface: eth1
# override.
iface: '{{ ansible_default_ipv4.interface }}'
ha_enabled: False
#Variables for <core-site_xml> - common
fs_default_FS_port: 8020
nameservice_id: mycluster4
#Variables for <hdfs-site_xml>
dfs_permissions_superusergroup: hdfs
- /namedir1/
- /namedir2/
dfs_replication: 3
dfs_namenode_handler_count: 50
dfs_blocksize: 67108864
- /datadir1/
- /datadir2/
dfs_datanode_address_port: 50010
dfs_datanode_http_address_port: 50075
dfs_datanode_ipc_address_port: 50020
dfs_namenode_http_address_port: 50070
dfs_ha_zkfc_port: 8019
qjournal_port: 8485
qjournal_http_port: 8480
dfs_journalnode_edits_dir: /journaldir/
zookeeper_clientport: 2181
zookeeper_leader_port: 2888
zookeeper_election_port: 3888
#Variables for <mapred-site_xml> - common
mapred_job_tracker_ha_servicename: myjt4
mapred_job_tracker_http_address_port: 50030
mapred_task_tracker_http_address_port: 50060
mapred_job_tracker_port: 8021
mapred_ha_jobtracker_rpc-address_port: 8023
mapred_ha_zkfc_port: 8018
mapred_job_tracker_persist_jobstatus_dir: /jobdir/
- /mapred1/
- /mapred2/
hadoop1 zoo_id=1
hadoop2 zoo_id=2
hadoop3 zoo_id=3
# Launch Job to count occurance of a word.
- hosts: $server
user: root
- name: copy the file
copy: src=inputfile dest=/tmp/inputfile
- name: upload the file
shell: su - hdfs -c "hadoop fs -put /tmp/inputfile /inputfile"
- name: Run the MapReduce job to count the occurance of word hello
shell: su - hdfs -c "hadoop jar /usr/lib/hadoop-0.20-mapreduce/hadoop-examples.jar grep /inputfile /outputfile 'hello'"
- name: Fetch the outputfile to local tmp dir
shell: su - hdfs -c "hadoop fs -get /outputfile /tmp/outputfile"
- name: Get the outputfile to ansible server
fetch: dest=/tmp src=/tmp/outputfile/part-00000
name=Cloudera's Distribution for Hadoop, Version 4
gpgkey =
gpgcheck = 1
- name: restart iptables
service: name=iptables state=restarted
# The playbook for common tasks
- name: Deploy the Cloudera Repository
copy: src=etc/cloudera-CDH4.repo dest=/etc/yum.repos.d/cloudera-CDH4.repo
- name: Install the libselinux-python package
yum: name=libselinux-python state=installed
- name: Install the openjdk package
yum: name=java-1.6.0-openjdk state=installed
- name: Create a directory for java
file: state=directory path=/usr/java/
- name: Create a link for java
file: src=/usr/lib/jvm/java-1.6.0-openjdk- state=link path=/usr/java/default
- name: Create the hosts file for all machines
template: src=etc/hosts.j2 dest=/etc/hosts
- name: Disable SELinux in conf file
selinux: state=disabled
- name: Create the iptables file for all machines
template: src=iptables.j2 dest=/etc/sysconfig/iptables
notify: restart iptables
# The playbook for common tasks
- include: common.yml tags=slaves localhost
{% for host in groups.all %}
{{ hostvars[host]['ansible_' + iface].ipv4.address }} {{ host }}
{% endfor %}
<?xml version="1.0"?>
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
See the License for the specific language governing permissions and
limitations under the License.
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<value>hdfs://{{ hostvars[groups['hadoop_masters'][0]]['ansible_hostname'] + ':' ~ hadoop['fs_default_FS_port'] }}/</value>
# Configuration of the "dfs" context for null
# Configuration of the "dfs" context for file
# Configuration of the "dfs" context for ganglia
# Pick one: Ganglia 3.0 (former) or Ganglia 3.1 (latter)
# dfs.class=org.apache.hadoop.metrics.ganglia.GangliaContext
# dfs.class=org.apache.hadoop.metrics.ganglia.GangliaContext31
# dfs.period=10
# dfs.servers=localhost:8649
# Configuration of the "mapred" context for null
# Configuration of the "mapred" context for file
# Configuration of the "mapred" context for ganglia
# Pick one: Ganglia 3.0 (former) or Ganglia 3.1 (latter)
# mapred.class=org.apache.hadoop.metrics.ganglia.GangliaContext
# mapred.class=org.apache.hadoop.metrics.ganglia.GangliaContext31
# mapred.period=10
# mapred.servers=localhost:8649
# Configuration of the "jvm" context for null
# Configuration of the "jvm" context for file
# Configuration of the "jvm" context for ganglia
# jvm.class=org.apache.hadoop.metrics.ganglia.GangliaContext
# jvm.class=org.apache.hadoop.metrics.ganglia.GangliaContext31
# jvm.period=10
# jvm.servers=localhost:8649
# Configuration of the "rpc" context for null