A New Data Layout Scheme for Energy-Efficient MapReduce Processing Tasks

Yet Another Resource Negotiator (YARN) is a framework to manage and allocate resource requests from applications that process big data stored in HDFS. However, dynamic power management methods are not efficient when YARN manage applications to process big data stored in the default data layout of HDFS. In this paper, we propose a new data layout scheme that can be implemented for HDFS. A comparison between our proposal and the existing HDFS data layout scheme shows that the new data layout algorithm significantly reduces the energy consumption at the slight expense of the mean response time of jobs.

The research has been partially supported by the European Union, co-financed by the European Social Fund (EFOP-3.6.2-16-2017-00013).

This research was partially supported by Basic Science Research Program through the National Research Foundation of Korea(NRF) funded by the Ministry of Science, ICT & Future Planning (2017R1A2B4009410).

The authors are grateful for anonymous reviewers’ and guest editors’ comments that helped to improve the quality of the paper.

Authors and Affiliations

1. Division of Knowledge and System Engineering for ICT, Faculty of Information Technology, Ton Duc Thang University, Ho Chi Minh City, Vietnam

   Tien Van Do

2. Department of Networked Systems and Services, Budapest University of Technology and Economics, H-117, Magyar tudósok körútja 2., Budapest, Hungary

   Xuan T. Tran & Tien Van Do

3. Nokia Bell Labs, Bókay János u. 36-42, 1082, Budapest, Hungary

   Csaba Rotter

4. Department of Computer Engineering, Yeungnam University, Gyeongsan, Korea

   Dosam Hwang

Corresponding author

Correspondence to Tien Van Do.

Appendix A: The Operation of HDFS and YARN in a Computing Cluster

1.1 A.1 HDFS

In a shared cluster, Hadoop Distributed File System (HDFS) [9] offers big data services for applications. The master NameNode (JVM process) manages the file metadata, and DataNodes (JVM processes) keep data in the form of data blocks/chunks. To read/write a file, an HDFS client initiates a request to the NameNode for metadata, then directly opens TCP sessions to specific DataNodes to stream data [25].

1.2 A.2 Yet Another Resource Negotiator – YARN

The Hadoop YARN framework [8, 26] orchestrates the resource management and allocation for applications with the use of three core Java software components: a central scheduler - ResourceManager (RM), per-node NodeManager (NM), and application-specific ApplicationMaster (AM).

• The master ResourceManager (RM) manages a global view of the cluster resources by processing the heartbeat messages from the NodeManagers and makes decisions about resource allocation based on the negotiation with an application-specific ApplicationMaster.

• A per-node NodeManager (denoted by NM _{i j} if it is hosted in server (i,j) as illustrated in Fig. 1) holds information related to the host (e.g. hostname, node label, rack position, available memory, CPUs, etc). It is responsible for heartbeating to RM with information about available resources in the host and launching containers for job processing. A container (aka YarnChild java process) runs a task.

• An application-specific ApplicationMaster (AM) (runs in a container) initiates resource requests for the application and manages all aspects of the job during its runtime.

1.3 A.3 Processing Hadoop MapReduce Applications

A Hadoop MapReduce application [27] is composed of a set of Map tasks and Reduce tasks. Map tasks read and process data blocks stored in HDFS, while Reduce tasks process Map outputs and report final.

In what follows (also Fig. 13), the cooperation of HDFS and YARN components is illustrated through the execution of a MapReduce Wordcount job that runs on YARN to count the occurrences of words in a file of two data blocks.

• When the application is submitted, a job client (JVM process) is created and communicates with the ResourceManager (RM) for launching the job (action (1)) (asking the application id and requesting to launch its MRAppMaster-the ApplicationMaster of the MapReduce framework).

• The RM grants a resource capacity for the MRAppMaster, then contacts with the respective NM (action (2.1)), which in turn launches an MRAppMaster container in the host (action (2.2)).

• The MRAppMaster registers itself with the RM (action (3.1)), contacts the NameNode (NN) to get the input file’s metadata (map split numbers, locations of blocks) (action (3.2)).

• The MRAppMaster constructs resource requests and sends to the RM. Based on the cluster available resources and scheduling mechanism, the RM responds the MRAppMaster with an allocation (action (4)). An allocation can contain one or many granted containers. The MRAppMaster iterates resource requests to the RM until all tasks (two Map tasks and one Reduce task) are computed.

• With a received allocation, the MRAppMaster asks respective NM for spawning containers (action (5.1)). A called NodeManager, in turn, creates an isolated YarnChild JVM process (action (5.2)) for task execution.

• Each Map task, based on the split’s metadata, opens a TCP/IP connection to a certain DataNode to read the data block (action (6)), does computation, generates and stores Map output on the local disk. The Reduce task, otherwise, reads the Map outputs and apply a reduce function over them for the final output, which is written back to HDFS without replication.

• During the runtime, Map/Reduce tasks directly report their status to the MRAppMaster (action (7)).

• After the Reduce task completes, the MRAppMaster deregisters itself with the RM (action (8)).

An example of MapReduce WordCount execution on a Hadoop computing cluster

Full size image

1.4 A.4 The Default HDFS Data Layout

NameNode daemon (runs inside a Java Virtual Machine) manages the filesystem tree, the metadata of all the files and directories, while DataNodes store blocks of files. Let DN _{i j} denote the DataNode daemon in server (i,j) (\(i = 1,\dots , K\), \(j = 1,\dots , M(i)\)), as shown in Fig. 1.

When a client submits the write request of a file (with information of type, file size and block size), the Namenode registers the file in its file system tree. For each block with size of BS Megabytes, the NameNode gives the list of RF DataNodes, where RF is the replication factor.

For RF = 3, the selection of DataNodes is as follows (see Algorithm 2):

• the first replica is stored on the local node if the node is inside the cluster or on a random node if the writer runs outside (the local node does not host DataNode);

• the second replica is on a random node at a different rack; and

• the third is on another node in the same rack with the second one.

After receiving a list of three DataNodes for a block, the client opens a TCP pipeline to the first DataNode in the list. Data is split into 4KB pieces to be flushed from the local temporary file to the first DataNode, which will transfer data pieces to the second and then to the next. As a next piece can be transferred before the acknowledgement for the completion of the previous piece, writing replicas of the same block is roughly concurrent in a write time of t _{b l k} seconds.

1.5 A.5 A Locality Relaxation Algorithm for Resource Allocation in RM

In YARN, resource allocation for an application is done through a negotiation between the ApplicationMaster (AM) and the ResourceManager (RM) and decided by the RM (action (4) in the example of WordCount in Fig. 13). When AM constructs a resource request (containing the request priority, preferred node, resource capacity per container, and the number of expected containers), data locality is taken account to reduce network cost. I.e., assuming DataNode DN _{i j} holds one or more data splits, local-data NodeManager NM _{i j} is preferred.

The RM scheduler uses the information of a resource request and available resources for the assignment of tasks to servers as follow:

• First, the RM tries to pick up the data-local node NM _{i j} (RM calculates the available resources of node NM _{i j} and checks if it meets the job’s resource requirement.

• If failed, it starts a container on another node in the same rack i, or on any node in the cluster (off-rack) if still unsuccessfully.

• If no resource is enough to start a container, a task must wait in the queue.

Let T _M and T _R denote the number of Map tasks and the number of Reduce tasks of a MapReduce job; N _{i j} and RAM _{i j} be the total cores and memory (GB) of NodeManager NM _{i j} ; FreeN _{i j} (0 ≤ FreeN _{i j} ≤ N _{i j} ) and FreeRAM _{i j} (0 ≤ FreeRAM _{i j} ≤ RAM _{i j} ) be the available cores and GB RAM amount on NM _{i j} at the current instant, respectively. Assuming that a container requires resource capacity of n core and m GB RAM, the RM resource allocation is illustrated in Algorithm 3.

Reprints and permissions