Administering HDFS

Cluster Maintenance

You can decommission slave nodes, manually add slave nodes to a cluster, and use DistCp to copy files between clusters.

Decommissioning Slave Nodes

Hadoop provides the decommission feature to retire a set of existing slave nodes (DataNodes, NodeManagers, or HBase RegionServers) in order to prevent data loss.

Slaves nodes are frequently decommissioned for maintenance. As a Hadoop administrator, you will decommission the slave nodes periodically in order to either reduce the cluster size or to gracefully remove dying nodes.

Prerequisites to decommission slave nodes

Make sure the appropriate properties are defined in the hdfs-site.xml file and yarn-site.xml files.

• Ensure that the following property is defined in your hdfs-site.xml file.

<property>

​

<name>dfs.hosts.exclud</name>  

​

<value>[HADOOP_CONF_DIR]/dfs.exclude</value>

​

</property>

where <HADOOP_CONF_DIR> is the directory for storing the Hadoop configuration files. For example, /etc/ hadoop/conf.

• Ensure that the following property is defined in your yarn-site.xml file.

<property>

​

<name>yarn.resourcemanager.nodes.exclude-path</name>  

​

<value>[HADOOP_CONF_DIR]/yarn.exclude</value>

​

<final>true</final> 

​

</property>

where <HADOOP_CONF_DIR> is the directory for storing the Hadoop configuration files. For example, /etc/ hadoop/conf.

Decommission DataNodes or NodeManagers

Nodes normally run both a DataNode and a NodeManager, and both are typically commissioned or decommissioned together.

With the replication level set to three, HDFS is resilient to individual DataNodes failures. However, there is a high chance of data loss when you terminate DataNodes without decommissioning them first. Nodes must be decommissioned on a schedule that permits replication of blocks being decommissioned.

On the other hand, if a NodeManager is shut down, the ResourceManager will reschedule the tasks on other

nodes in the cluster. However, decommissioning a NodeManager may be required in situations where you want a NodeManager to stop to accepting new tasks, or when the tasks take time to execute but you still want to be agile in your cluster management.

Decommission DataNodes

Edit the configuration files and execute commands on the NameNode host machine.

About this task

Use the following instructions to decommission DataNodes in your cluster:

Procedure

1. On the NameNode host machine, edit the <HADOOP_CONF_DIR>/dfs.exclude file and add the list of DataNodes hostnames (separated by a newline character).

where <HADOOP_CONF_DIR> is the directory for storing the Hadoop configuration files. For example, /etc/ hadoop/conf.

2. Update the NameNode with the new set of excluded DataNodes. On the NameNode host machine, execute the following command:

su <HDFS\_USER>

​

hdfs dfsadmin -refreshNodes

where <HDFS_USER> is the user owning the HDFS services. For example, hdfs.

3. Open the NameNode web UI (http://<NameNode_FQDN>:50070) and navigate to the DataNodes page. Check to see whether the state has changed to Decommission In Progress for the DataNodes being decommissioned.
4. When all the DataNodes report their state as Decommissioned (on the DataNodes page, or on the Decommissioned Nodes page at http://<NameNode_FQDN>:8088/cluster/ nodes/decommissioned), all of the blocks have been replicated. You can then shut down the decommissioned nodes.
5. If your cluster utilizes a dfs.include file, remove the decommissioned nodes from the <HADOOP_CONF_DIR>/ dfs.include file on the NameNode host machine, then execute the following command:

su <HDFS\_USER> 

​

hdfs dfsadmin -refreshNodes

Decommission NodeManagers

Edit the configuration files on the ResourceManager host machine and add the details of the new set of NodeManagers.

About this task

Use the following instructions to decommission NodeManagers in your cluster:

Procedure

1. On the ResourceManager host machine, edit the <HADOOP_CONF_DIR>/yarn.exclude file and add the list of NodeManager hostnames (separated by a newline character).

where <HADOOP_CONF_DIR> is the directory for storing the Hadoop configuration files. For example, /etc/ hadoop/conf.

2. If your cluster utilizes a yarn.include file, remove the decommissioned nodes from the <HADOOP_CONF_DIR>/ yarn.include file on the ResourceManager host machine.

Note If no yarn.include file is specified, all NodeManagers are considered to be included in the cluster (unless excluded in the yarn.exclude file). The yarn.resourcemanager.nodes.include-path and yarn.resourcemanager.nodes.exclude-path properties in yarn-site.xml are used to specify the yarn.include and yarn.exclude files.

3. Update the Resource Manager with the new set of NodeManagers. On the ResourceManager host machine, execute the following command:

su <YARN\_USER>

​

yarn rmadmin -refreshNodes

where <YARN_USER> is the user who owns the YARN services, for example, yarn.

Decommission HBase Region Servers

Run the decommission commands on the RegionServer.

About this task

Use the following instruction to decommission HBase RegionServers in your cluster.

At the RegionServer that you want to decommission, run the following command:

su <HBASE\_USER> 

​

/usr/odp/current/hbase-client/bin/hbase-daemon.sh stop

where <HBASE_USER> is the user who owns the HBase Services. For example, hbase. RegionServer closes all the regions, then shuts down.

Manually Add Slave Nodes to an ODP Cluster

You can add slave nodes as well as the HBase RegionServer. In this section:

• Slave node prerequisites
• Add slave nodes
• Add HBase RegionServer

Prerequisites to Manually Add Slave Nodes

Make sure the ports are available, the database is deployed, and the correct JDK version is installed on all the nodes in the cluster.

Ensure that the new slave nodes meet the following prerequisites:

• At each of your hosts:
• yum (RHEL)
• zypper
• rpm
• scp
• curl
• wget
• unzip
• tar
• pdsh
• Ensure that all of the ports are available.
• To install Hive metastore or to use an external database for Oozie metastore, ensure that you deploy either a MariaDB, MySQL or an Oracle database in your cluster.
• Your system must have the correct JDK installed on all of the nodes in the cluster.

Add Slave Nodes

Configure the remote repository, install HDFS, compression libraries, and copy the Hadoop configurations to the new slave nodes.

About this task

Use the following instructions to manually add a slave node:

Procedure

1. On each new slave node, configure the remote repository.
2. On each new slave node, install HDFS.
3. On each new slave node, install compression libraries.
4. On each new slave node, create the DataNode and YARN NodeManager local directories.
5. Copy the Hadoop configurations to the new slave nodes and set appropriate permissions.

• Option I: Copy Hadoop config files from an existing slave node.
• On an existing slave node, make a copy of the current configurations:

tar zcvf hadoop_conf.tgz /etc/hadoop/conf

• Copy this file to each of the new nodes:

rm -rf /etc/hadoop/conf

​

cd /

​

tar zxvf $location_of_copied_conf_tar_file/hadoop_conf.tgz 

​

chmod -R 755 /etc/hadoop/conf

6. On each of the new slave nodes, start the NodeManager:

su -l yarn -c "/usr/odp/current/hadoop-yarn-nodemanager/sbin/yarn-daemon.sh start nodemanager"

7. Optional - If you use a HDFS or YARN/ResourceManager .include file in your cluster, add the new slave nodes to the .include file, then run the applicable refreshNodes command.

• To add new DataNodes to the dfs.include file:
• On the NameNode host machine, edit the /etc/hadoop/conf/dfs.include file and add the list of the new slave node host names (separated by newline character).

Note If no dfs.include file is specified, all DataNodes are considered to be included in the cluster (unless excluded in the dfs.exclude file). The >dfs.hosts and dfs.hosts.exlude properties in hdfs-site.xml are used to specify the dfs.include and dfs.exclude files.

• On the NameNode host machine, execute the following command:

su -l hdfs -c "hdfs dfsadmin -refreshNodes"

• To add new NodeManagers to the yarn.include file:
• On the ResourceManager host machine, edit the /etc/hadoop/conf/yarn.include file and add the list of the slave node host names (separated by newline character).

Note If no yarn.include file is specified, all NodeManagers are considered to be included in the cluster (unless excluded in the yarn.exclude file). The >yarn.resourcemanager.nodes.include-path and yarn.resourcemanager.nodes.exclude-path properties in yarn-site.xml are used to specify the yarn.include >and yarn.exclude files.

• On the ResourceManager host machine, execute the following command:

su -l yarn -c "yarn rmadmin -refreshNodes"

Add HBase Region Server

Install HBase, ZooKeeper, add the ODP repository, and copy the HBase configurations to the new slave nodes.

About this task

Use the following instructions to manually add HBase RegionServer hosts:

Procedure

1. On each of the new slave nodes, install HBase and ZooKeeper.

• For RHEL/CentOS

yum install zookeeper hbase

2. On each of the new slave nodes, add the ODP repository to yum:

• RHEL/CentOS:

wget -nv http://public-repo-1.odp.com/ODP/centos6/2.x/updates/2.3.0.0/odp.repo -O /etc/yum.repos.d/odp.repo

3. Copy the HBase configurations to the new slave nodes and set appropriate permissions.

• Option I: Copy HBase config files from an existing slave node.
• On any existing slave node, make a copy of the current configurations:

tar zcvf hbase_conf.tgz /etc/hbase/conf

​

tar zcvf zookeeper_conf.tgz /etc/zookeeper/conf

• Copy these files to each of the new nodes:

rm -rf /etc/hbase/conf

​

mkdir -p /etc/hbase/conf

​

cd /

​

tar zxvf $location_of_copied_conf_tar_file/hbase_conf.tgz

​

​

​

chmod -R 755 /etc/hbase/conf

​

rm -rf /etc/zookeeper/conf

​

mkdir -p /etc/zookeeper/conf

​

cd /

​

tar zxvf $location_of_copied_conf_tar_file/zookeeper_conf.tgz 

​

chmod -R 755 /etc/zookeeper/conf

• Option II: Manually add Hadoop configuration files.

4. On all of the new slave nodes, create the configuration directory, copy all of the configuration files, and set the permissions:

rm -r $HBASE\_CONF\_DIR ; 

mkdir -p $HBASE\_CONF\_DIR ;

Copy all of the configuration files to $HBASE_CONF_DIR

chmod a+x $HBASE_CONF_DIR/;

​

chown -R $HBASE_USER:$HADOOP_GROUP $HBASE_CONF_DIR/../ ; 

​

chmod -R 755 $HBASE_CONF_DIR/../

rm -r $ZOOKEEPER_CONF_DIR ; 

​

mkdir -p $ZOOKEEPER_CONF_DIR

Copy all of the configuration files to $ZOOKEEPER_CONF_DIR

chmod a+x $ZOOKEEPER_CONF_DIR/;

​

chown -R $ZOOKEEPER_USER:$HADOOP_GROUP $ZOOKEEPER_CONF_DIR/../ ; 

​

chmod -R 755 $ZOOKEEPER_CONF_DIR/../

where:

• $HBASE_CONF_DIR is the directory to store the HBase configuration files. For example, /etc/hbase/conf.
• $HBASE_USER is the user owning the HBase services. For example, hbase.
• $HADOOP_GROUP is a common group shared by services. For example, hadoop.
• $ZOOKEEPER_CONF_DIR is the directory to store the ZooKeeper configuration files. For example, /etc/ zookeeper/conf
• $ZOOKEEPER_USER is the user owning the ZooKeeper services. For example, zookeeper.

5. Start HBase RegionServer node:

<login as $HBASE_USER>

​

/usr/lib/hbase/bin/hbase-daemon.sh --config $HBASE_CONF_DIR start regionserver

6. On the HBase Master host machine, edit the /usr/lib/hbase/conf file and add the list of slave nodes' hostnames. The hostnames must be separated by a newline character.

Using DistCp to Copy Files

Hadoop DistCp (distributed copy) can be used to copy data between Hadoop clusters (and also within a Hadoop cluster).

DistCp uses MapReduce to implement its distribution, error handling, and reporting. It expands a list of files and directories into map tasks, each of which copies a partition of the files specified in the source list.

Using DistCp

Use DistCp to copy files between various clusters.

The most common use of DistCp is an inter-cluster copy:

hadoop distcp hdfs://nn1:8020/source hdfs://nn2:8020/destination

Where hdfs: //nn1:8020/source is the data source, and hdfs://nn2:8020/ destination is the destination. This will expand the name space under /source on NameNode "nn1" into a temporary file, partition its contents among a set of map tasks, and start copying from "nn1" to "nn2". Note that DistCp requires absolute paths.

You can also specify multiple source directories:

hadoop distcp hdfs://nn1:8020/source/a hdfs://nn1:8020/source/b hdfs://nn2:8020/destination

Or specify multiple source directories from a file with the -f option:

hadoop distcp -f hdfs://nn1:8020/srclist hdfs://nn2:8020/destination

Where srclist contains:

hdfs://nn1:8020/source/a 

hdfs://nn1:8020/source/b

When copying from multiple sources, DistCp will abort the copy with an error message if two sources collide, but collisions at the destination are resolved based on the options specified. By default, files already existing at the destination are skipped (i.e. not replaced by the source file). A count of skipped files is reported at the end of each job, but it may be inaccurate if a copier failed for some subset of its files, but succeeded on a later attempt.

It is important that each NodeManager is able to communicate with both the source and destination file systems. For HDFS, both the source and destination must be running the same version of the protocol, or use a backwards- compatible protocol; see "Copying Between Versions".

After a copy, you should generate and cross-check a listing of the source and destination to verify that the copy was truly successful. Since DistCp employs both Map/Reduce and the FileSystem API, issues in or between any of these three could adversely and silently affect the copy. Some have had success running with -update enabled to perform a second pass, but users should be acquainted with its semantics before attempting this.

It is also worth noting that if another client is still writing to a source file, the copy will likely fail. Attempting to overwrite a file being written at the destination should also fail on HDFS. If a source file is (re)moved before it is copied, the copy will fail with a FileNotFound exception.

Command Line Options

You can use command line operations to perform DistCp copying.

For a description of DistCp command line options, see DistCp Command Line Options.

Update and Overwrite

Use the -update option to copy files from a source when they do not exist at the target. Use the -overwrite function to overwrite the target files even if the content is the same.

The DistCp -update option is used to copy files from a source that does not exist at the target, or that has different contents. The DistCp -overwrite option overwrites target files even if they exist at the source, or if they have the same contents.

The -update and -overwrite options warrant further discussion, since their handling of source-paths varies from the defaults in a very subtle manner.

Consider a copy from /source/first/ and /source/second/ to /target/, where the source paths have the following contents:

hdfs://nn1:8020/source/first/1 hdfs://nn1:8020/source/first/2 hdfs://nn1:8020/source/second/10 hdfs://nn1:8020/source/second/20

When DistCp is invoked without -update or -overwrite, the DistCp defaults would create directories first/ and second/, under /target. Thus:

distcp hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

would yield the following contents in /target:

hdfs://nn2:8020/target/first/1 hdfs://nn2:8020/target/first/2 hdfs://nn2:8020/target/second/10 hdfs://nn2:8020/target/second/20

When either -update or -overwrite is specified, the contents of the source directories are copied to the target, and not the source directories themselves. Thus:

distcp -update hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second  hdfs://nn2:8020/target

would yield the following contents in /target:

hdfs://nn2:8020/target/1 hdfs://nn2:8020/target/2 hdfs://nn2:8020/target/10 hdfs://nn2:8020/target/20

By extension, if both source folders contained a file with the same name ("0", for example), then both sources would map an entry to /target/0 at the destination. Rather than permit this conflict, DistCp will abort.

Now, consider the following copy operation:

distcp hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

With sources/sizes:

hdfs://nn1:8020/source/first/1 32 hdfs://nn1:8020/source/first/2 hdfs://nn1:8020/source/second/10 64 hdfs://nn1:8020/source/second/20

And destination/sizes:

hdfs://nn2:8020/target/1 32 

hdfs://nn2:8020/target/10 32 

hdfs://nn2:8020/target/20 64

Will effect:

hdfs://nn2:8020/target/1 32

hdfs://nn2:8020/target/2 32 

hdfs://nn2:8020/target/10 64 

hdfs://nn2:8020/target/20 32

1 is skipped because the file-length and contents match. 2 is copied because it does not exist at the target. 10 and 20 are overwritten because the contents don’t match the source.

If the -update option is used, 1 is overwritten as well.

DistCp and Security Settings

Security settings dictate whether DistCp should be run on the source cluster or the destination cluster.

The general rule-of-thumb is that if one cluster is secure and the other is not secure, DistCp should be run from the secure cluster -- otherwise there may be security- related issues.

When copying data from a secure cluster to an non-secure cluster, the following configuration setting is required for the DistCp client:

<property>

​

<name>ipc.client.fallback-to-simple-auth-allowed</name>  

​

<value>true</value>

​

</property>

When copying data from a secure cluster to a secure cluster, the following configuration setting is required in the core-site.xml file:

<property>

​

<name>hadoop.security.auth_to_local</name>

​

<value></value>

​

<description>Maps kerberos principals to local user names</description> 

​

</property>

DistCp between HA cluster

To copy data between HA clusters, use the dfs.internal.nameservices property in the hdfs-site.xml file to explicitly specify the name services belonging to the local cluster, while continuing to use the dfs.nameservices property to specify all of the name services in the local and remote clusters.

About this task

Use the following steps to copy data between HA clusters:

Modify the following properties in the hdfs-site.xml file for both cluster A and cluster B:

Procedure

1. Add both name services to dfs.nameservices = HAA, HAB
2. Add the dfs.internal.nameservices property:

• In cluster A:

dfs.internal.nameservices = HAA

• In cluster B:

dfs.internal.nameservices = HAB

3. Add dfs.ha.namenodes. to both clusters:

• In cluster A

dfs.ha.namenodes.HAB = nn1,nn2

• In cluster B

dfs.ha.namenodes.HAA = nn1,nn2

4. Add the dfs.namenode.rpc-address.. property:

• In Cluster A:

dfs.namenode.rpc-address.HAB.nn1 = <NN1_fqdn>:8020 dfs.namenode.rpc-address.HAB.nn2 = <NN2_fqdn>:8020

• In Cluster B:

dfs.namenode.rpc-address.HAA.nn1 = <NN1_fqdn>:8020 dfs.namenode.rpc-address.HAA.nn2 = <NN2_fqdn>:8020

5. Add the following properties to enable distcp over WebHDFS and secure WebHDFS:

• In Cluster A:

dfs.namenode.http-address.HAB.nn1 = <NN1_fqdn>:50070 dfs.namenode.http-address.HAB.nn2 = <NN2_fqdn>:50070 dfs.namenode.https-address.HAB.nn1 = <NN1_fqdn>:50470 dfs.namenode.https-address.HAB.nn2 = <NN2_fqdn>:50470

• In Cluster B:

dfs.namenode.http-address.HAA.nn1 = <NN1_fqdn>:50070 dfs.namenode.http-address.HAA.nn2 = <NN2_fqdn>:50070 dfs.namenode.https-address.HAA.nn1 = <NN1_fqdn>:50470 dfs.namenode.https-address.HAA.nn2 = <NN2_fqdn>:50470

6. Add the dfs.client.failover.proxy.provider. property:

• In cluster A:

dfs.client.failover.proxy.provider. HAB = org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider

• In cluster B:

dfs.client.failover.proxy.provider. HAA = org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider

7. Restart the HDFS service, then run the distcp command using the NameService. For example: hadoop distcp hdfs://HAA/tmp/testDistcp hdfs://HAB/tmp/

DistCp Frequently Asked Questions

There are differences between DistCp latest version and the legacy DistCp versions.

• Why does -update not create the parent source directory under a pre-existing target directory? The behavior of - update and -overwrite is described in detail in the Using DistCp section of this document. In short, if either option is used with a pre-existing destination directory, the contents of each source directory are copied over, rather than the source directory itself. This behavior is consistent with the legacy DistCp implementation.

• How does the new DistCp (version 2) differ in semantics from the legacy DistCp?

  • Files that are skipped during copy previously also had their file-attributes (permissions, owner/group info, etc.) unchanged, when copied with Legacy DistCp. These are now updated, even if the file copy is skipped.
  • In Legacy DistCp, empty root directories among the source path inputs were not created at the target. These are now created.

• Why does the new DistCp (version 2) use more maps than legacy DistCp? Legacy DistCp works by figuring out what files need to be actually copied to target before the copy job is launched, and then launching as many maps as required for copy. So if a majority of the files need to be skipped (because they already exist, for example), fewer maps will be needed. As a consequence, the time spent in setup (i.e. before the MapReduce job) is higher. The new DistCp calculates only the contents of the source paths. It does not try to filter out what files can be skipped. That decision is put off until the MapReduce job runs. This is much faster (vis-a-vis execution-time), but the number of maps launched will be as specified in the -m option, or 20 (the default) if unspecified.

• Why does DistCp not run faster when more maps are specified? At present, the smallest unit of work for DistCp is a file. i.e., a file is processed by only one map. Increasing the number of maps to a value exceeding the number of files would yield no performance benefit. The number of maps launched would equal the number of files.

• Why does DistCp run out of memory? If the number of individual files/directories being copied from the source path(s) is extremely large (e.g. 1,000,000 paths), DistCp might run out of memory while determining the list of paths for copy. This is not unique to the new DistCp implementation. To get around this, consider changing the - Xmx JVM heap- size parameters, as follows:

bash$ export HADOOP_CLIENT_OPTS="-Xms64m -Xmx1024m" 

bash$ hadoop distcp /source /target

DistCp Additional Considerations

DistCp also provides a strategy to “dynamically” size maps, allowing faster DataNodes to copy more bytes than slower nodes.

Map Sizing

By default, DistCp makes an attempt to size each map comparably so that each copies roughly the same number of bytes. Note that files are the finest level of granularity, so increasing the number of simultaneous copiers (i.e. maps) may not always increase the number of simultaneous copies nor the overall throughput.

Using the dynamic strategy (explained in the Architecture), rather than assigning a fixed set of source files to each map task, files are instead split into several sets. The number of sets exceeds the number of maps, usually by a factor of 2-3. Each map picks up and c opies all files listed in a chunk. When a chunk is exhausted, a new chunk is acquired and processed, until no more chunks remain.

By not assigning a source path to a fixed map, faster map tasks (i.e. DataNodes) are able to consume more chunks -- and thus copy more data -- than slower nodes. While this distribution is not uniform, it is fair with regard to each mapper’s capacity.

The dynamic strategy is implemented by the DynamicInputFormat. It provides superior performance under most conditions.

Tuning the number of maps to the size of the source and destination clusters, the size of the copy, and the available bandwidth is recommended for long-running and regularly run jobs.

Copying Between Versions of HDFS

For copying between two different versions of Hadoop, you will usually use HftpFileSystem. This is a read-only FileSystem, so DistCp must be run on the destination cluster (more specifically, on NodeManagers that can write to the destination cluster). Each source is specified as hftp://<dfs.http.address>/<path> (the default dfs.http.address is <namenode>:50070 ).

MapReduce and Other Side-Effects

As mentioned previously, should a map fail to copy one of its inputs, there will be several side-effects.

• Unless -overwrite is specified, files successfully copied by a previous map will be marked as “skipped” on a re- execution.
• If a map fails mapreduce.map.maxattempts times, the remaining map tasks will be killed (unless -i is set).
• If mapreduce.map.speculative is set final and true, the result of the copy is undefined.

SSL Configurations for HSFTP Sources

To use an HSFTP source (i.e. using the HSFTP protocol), a SSL configuration file needs to be specified (via the - mapredSslConf option). This must specify 3 parameters:

• ssl.client.truststore.location: The local file system location of the trust-store file, containing the certificate for the NameNode.
• ssl.client.truststore.type: (Optional) The format of the trust-store file.
• ssl.client.truststore.password: (Optional) Password for the trust-store file.

The following is an example of the contents of a SSL Configuration file:

<configuration>

​

<property>

​

<name>ssl.client.truststore.location</name>

​

<value>/work/keystore.jks</value>

​

<description>Truststore to be used by clients like distcp. Must be specified.</description>

​

</property>

​

<property>

​

<name>ssl.client.truststore.password</name>

​

<value>changeme</value>

​

<description>Optional. Default value is "".</description> </property>

​

<property>

​

<name>ssl.client.truststore.type</name>

​

<value>jks</value>

​

<description>Optional. Default value is "jks".</description>

​

</property>

​

</configuration>

The SSL configuration file must be in the classpath of the DistCp program.

Ports and Services Reference

Configure ports of various ODP services and control ODP services manually.