DEIM Forum 2014 A2-2

A Study of Many-Core Hardware Accelerated Hadoop MapReduce
Min Luo†Toshimori Honjo‡

Software Innovation Center, NTT Corporation
3-9-11 Midoricho, Musashino-shi, Tokyo, 180-8585 Japan
E-mail: †luo.min@lab.ntt.co.jp, ‡honjo.toshimori@lab.ntt.co.jp

Abstract MapReduce is a widely used framework for massive data processing. It was originally designed to overcome theI/O bottleneck, and enabled us to process Bigdata with the commodity clusters systems. However, several existing work haverecently shown that the emerging high speed storage and network devices are capable to remove the I/O bottleneck and madethe CPU the next serious bottleneck in the MapReduce framework. In this paper, we propose hardware accelerated (HA)
Hadoop MapReduce framework and implement it on a Tilera's many-core processor board to overcome the CPU bottleneck.
Our proposed solution offloads the main parts of Map and Reduce procedures, including the data parsing, sorting and merging.
Based on our experimental evaluations, we verify the feasibility of our proposal.
Keyword Hardware Acceleration，MapReduce，Many-core，Hadoop

1. Introduction

single node in commodity clusters, many solutions have

The explosion of information has created a new area of

been proposed including the compound use of GP-GPU

research for computer scientists, creatively named Big

and CPU [4], the hardware accelerators by many-core ordata. However, the limited HDD I/O performance is a

FPGA [5-9]. However, these works focused on the use of

typical bottleneck in processing the massive amount of big

accelerators to improve general system performance; there

data. Therefore, cluster-based parallel and distributed

is no offloading efficiency study or optimization for theapproach is proposed and achieves cost-effective total I/O

MapReduce framework on these accelerated proposals.

performance. Recently, there is popular cluster-based big

In light of the existing achievements from hardware

data processing application named Apache Hadoop [1] ,technology and their widely use in commodity servers inwhich contains an open source implementation of Google

near future, it is time to have a redesign of MapReduce

MapReduce framework [2]. It further relieves the data

framework and have a detailed performance study ofmoving overhead among servers by combining data

MapReduce phases on the hardware accelerated platforms.

storage and data computation into an integrated system,

In this paper, we propose our hardware accelerated

and enables massive data processing at high speed by the(H.A.)

commodity clusters systems. However, the data moving

implemented on a Tilera’s many-core processor board. In

overhead required in MapReduce framework still makes

our prototype system, the main parts of Map and Reduce

the limited HDD I/O speed in the commodity server andprocedures, including data parsing, sorting and merging,

the limited network bandwidth among servers the typicalare able to be offloaded. We verified the feasibility of ourperformance bottleneck in Hadoop.

proposal in overcoming the CPU bottlen eck through theOn the other hand, several kind of high speed storage

Hadoop

MapReduce

framework

which

is

benchmark experiments.

devices and networks have emerged recently, such as Solid

The remainder of the paper is organized as follows.

State Device (SSD) and Infiniband. The throughputs ofSection 2 gives the background of MapReduce frameworkthese devices are faster than those of conventional HDD

and its performance issue. Section 3 presents the design ofand connections by at least one order of magnitude. Our

our H.A. MapReduce framework, the offloading processes

previous research [3] has shown that these devices are

and implementation. Experimental benchmarks and theable to remove the previous bottlenecks in MapReduce

evaluation results are described in Section 4. Section 5cluster systems and verified the next bottleneck turns to be

discusses the related work on H.A. MapReduce. We

the limited CPU capability on the commodity cluster node.

conclude the lessons we learned and our future work in

In addition, to improve the calculation capability of a

Section 6.

2. An Introduction of MapReduce

the storage layer for its MapReduce computation model.
The HDFS consists of a single centralized management

2.1. MapReduce Framework

node, NameNode, which maintains all the metadata for the

MapReduce is a parallel programming model proposed

cluster along with multiple storage nodes; DataNodes,

by Google to facilitate large scale data processing in awhich contain all the data in large blocks size (default

distributed computing environment [2].

64MB). The MapReduce computation model includes a

The MapReduce’s programming model contains two

single central management node, JobTracker, and multiple

phases, map and reduce. As shown in Figure 1, the input

computation nodes, TaskTrackers. The JobTracker isdata to a computing task is split into many < K, V> tuples

responsible for initializing the specific map and reduceand a map function processes there pairs to generate a set

task for a submitted MapReduce job to a subset of theof intermediate <K’, V’> pairs. The intermediate pairs

TaskTrackers in the cluster. The JobTracker also monitorswith the same intermediate key are grouped together andthe TaskTrackers’ state to ensure redundancy and timely

passed to reduce function. The communication model

task completion. A single TaskTracker can be assigned

within MapReduce is transparent to users to alleviate theboth map and reduce tasks with the same job. Therefore,

development cost. MapReduce framework takes care of thethe computation and storage is able to be integrat ed

parallel execution by issuing and managing the map andbecause the DataNodes and TaskTrackers may exist on the

reduces tasks to computation nodes. It greatly relieves

same

the complexity of designing parallel computing programs

performance of MapReduce.

physical

node,

which

are

so

criticaltotheand provides efficiency for data-intensive applications [2].

There are five execution phases in Hadoop MapReduce.

There are many implementations of MapReduce model

When a job is submitted to the JobTracker, it divides theon parallel computing platforms in the last decade [ 1, 2,

input data into many data split and selects a number of

4-9].

TaskTracker and schedules th em to run several MapTasks,
one for each split.
In the first phase, the MapTask loads a chunk of its
input split data from HDFS to memory, de -serializes andparses it to extract key-value pairs. The mapping functionin each MapTask converts the original <K, V> pairs into a
set of intermediated results, which are records in the formof <K’, V’> pairs. The intermediate pairs are partitioned
and sorted by the <K’> values, and buffered in memory.
MapTask spills buffer content to disk based on a threshold

2.2. MapReduce Execution Overview
Because Apache Hadoop [1] is the most popular andmechanism in memory. It repeats the above sequence until
all its input split data is processed.

widely used open-source MapReduce implementation, we

In the second phase, MapTask merges all the spill files

focus our discussion on Hadoop ’s MapReduce hereafter.

on disk and generates a single MOF (Map Output File) to

Hadoop contains a network file system, named Hadoop

disk for its split data. Note that, this merge process may

Distributed File System (HDFS), which is implemented as

contain a ‘combine’ function in its final MOF generation

to reduce the data transfer overhead during the next phase.

gigabit Ethernet is now widely used, and 40 gigabit

In the third phase, the JobTracker selects a set of

Ethernet is available in the market. 100 Gigabit Ethernet

TaskTrackers

to

run

ReduceTasks

when

MOFs

are

will become common in the near future. Infiniband isavailable. Each ReduceTask fetches its intended data

another fascinating high speed network. Infiniband waspartition (also called segment) from the MOF across the

widely applied to super computers. Vendors are now

network once it has been generated by the MapTasks. This

offering Infiniband interface cards for commodity servers.

all-map-to-all-reduce phase is also named shuffle.

Current Infiniband network bandwidths are of the order of

The fourth phase starts after all the MOFs generationand shuffling is completed. The ReduceTask on each

56Gbit/s. In near future, these high speed devices will befound in commodity servers.

TaskTracker merges the shuffled segments into a single

Our previous research [3] has shown that these newfile, in which the data are partitioned and sorted by its key

devices eliminate the I/O bottleneck in MapReduce and<K’>. Specifically, because the segments are alreadythe next bottleneck in MapReduce turns to be the limited

sorted and the memory size is generally insufficient for all

CPU capability on the commodity cluster node.

the segments, segments only have a piece of theirfront-most data loaded in to memory for the comparison,sortingandmerging

processing.

Afterwriting

3. Proposed H.A. MapReduce framework

theprocessed data back to HDD, the following data pieces in

3.1. Hardware Acceleration in MapReduceeach segment will be loaded into memory and repeat the

There are many reasons why the clock frequency

above sequence until all the segments data are processed.

increase had been becoming saturated. The unsustainable

In the last phase, the merged results of every key <K’>level of power consumption implied by higher clock rates

in the fourth phase, as <K’, {V’ seg me nt _1 , V’ seg ment _ 2 , ...,

is one of the obvious and stringent reasons. Advances now

V’ seg me nt _n }>, is reduced to the final <K’, V’’> value pairconsist of increasing the number of cores. However, it

by the ReduceTask. This process is named reduce phase.

seems unlikely that number of host CPU cores will reachlevels sufficient to fully utilize the state-of-the-art storage

2.3. Bottlenecks for MapReduce Performance
In MapReduce execution, as described in Sec.2.2, bothand network devices. Some improvement in MapReduce
performance is possible by increasing the number ofthe map (map computing, map intermediate data splitting,servers,

and map merging) and reduce (reduce merging) processes

consumption neither.

require massive disk I/O operations.

butthis isinefficient

in

terms

of powerIn our solution, we extend the MapReduce framework

In addition, a reduce task will not start until all thewith a hardware accelerator board equipped with a special

MOFs are generated and all its corresponding segments inprocessor such as FP GA or many core processors. This

these MOFs are shuffled to the reduce node. In other word,

allows us to seamlessly increase the number of boards tothe shuffling phase may incur considerable performance

keep up with performance demands and fully utilize

overhead in MapReduce framework, especially for thestate-of-the-art storage and network devices. In addition,

shuffle-weight tasks (e.g., sorting, inverted -indexing).

as discussed in Sec. 2.2, the five main data processing

Note that the shuffle moves data from the map nodes ’ disk

phases consisted in MapReduce are isolated from eachrather than their memories, and through the cluster

other,

network. The affordable disk and network bandwidths ofaccelerating these phases individually.

commodity clusters incur great latency and become thebottleneck to the MapReduce performance [11].
However, several kinds of high speed storage andnetwork devices, such as Solid State Device (SSD) andwhich

logically

impliesthefeasibilityin

In this paper, we propose our H.A. MapReduce in whichdata mapping, data merging (including sorting) and data
reducing processing are able to be offloaded onto theaccelerators.

Infiniband, have emerged. While conventional HDD has atransfer-rate of about 100MB/s, SATA-based SSD offers

3.2. Prototype Implementation

500MB/s, and PCI Express-based SSD offers several GB/s.

We implement our prototype on a many core processor

In addition, it is possible to install a number of theseboard TILEncore-Gx36, which is developed by Tilera [12].

devices, so that each node can realistically offer theIt contains a cache-coherent mesh network of 36 tiles,

storage bandwidth of ~10GB/s. As for the network, 10

where each tile houses a general purpose processor, cacheIn our implementation, the accelerated merge phase
contains data sorting processing implicitly. It is suitable
for accelerating the merge phase in both MapTask andReduceTask. To offload the intermediate file merging
phase which happens at the end of a MapTask, Map ofHadoop MapReduce dispatches the MergeTasks to theoutside host process. This host process transforms the data
and a non-blocking router. Low power consumption isin the intermediate file assigned to each MergeTask into

exciting feature of this processor. The power consumption(key, value-pointer) format so as to reduce transfer amountof TILEncore-Gx36 is around 35W. One 1-GHz clocked

and memory usage on Tilera board. Multiple on-board

CPU (36cores) and 8GB RAM was implemented on theprocesses perform key sorting and value combining (if theacceleration board used in this experiment. The board has

user designed) in parallel for one MergeTask, and output

a PCI Express interface, and communicates with its hostthe merged file into host disk by using DMA transfer. Note

via this interface. The board runs SMP Linux so that wethat the returned merged file only contains (keys &can easily develop several applications in user mode

value-pointers), the values are filled into the merge file onNow, we give the overview of our map-offloading,
reduce-offloading and merge-offloading, respectively.
At first, we give the overview of our proposed H.A.

the host node. When all the intermediate files of a
MapTask are processed, the next MapReduce job isresumed. To offload the merge phase which happens at the

MapReduce framework. As shown in Figure 3, there are

beginningofaReduceTask,

ReduceofHadoopthree main parts consisted in our framework: Hadoop task,

MapReduce dispatches the MergeTasks to the acceleration

host process, and on-board process.

board in the similar way above. Note that the data to be

For map acceleration, when MapReduce executed, thesetransferred is the segments that are shuffled to current

offloaded tasks are launched on the accelerator board node.

ReduceTask’s host node. Because the records in these

We implemented a hook in Map of Hadoop MapReduce

segments are already sorted, we propose and implement asuch that calling it dispatches the tasks to the outside hostmulti-phaseprocess. This host process receives the data assigned toacceleration of Reduce-Merge more efficient.

the tasks, splits them into chunks, and transfers them tothe onboard process by using DMA transfer. The on -board
processuses

multi-thread

processing,andperforms

key-value pair generation and sort by key in 30 parallel
streams. The intermediate data is transferred back to thehost disk, and wait to be sorted into one merged file(MOF) during the merge offload phase. When all dataassigned in this MapTask is processed, control is returnedto the Map task, and the MapReduce job is resumed.

merging

strategy

to

make

the

mergingBy using this multi-phase merge strategy, as shown in
Figure 6, segments of a ReduceTask are sequentiallyloaded onto Tilera board. Every two loaded segments will
form a pair and be merged by a thread (core) in the highest
phase. The merged output data of these threads are stored
in a ring buffer. Threads of the lower phase will pull andmerge the upper phase output, and store their own merged
output in another ring buffer while releasing the memory
space of the upper ring buffer. Thus these multiple phases
form a pipeline so that all the Tilera cores are running inparallel to merge the segments. Note that a maximum of[n/2] threads can be issued in the first phase of our
implementation (n is the core number on Tilera board) .
In addition, we propose and implement an index

4. Experimental Evaluation
In this section, we will examine the effectiveness of our
H.A. MapReduce. We first demonstrate the multi-phasemerge efficiency, with “Terasort” benchmark. Then, we

structure for the final output file in the multi -phase merge

provideprocess. This index is useful in the reduce phase

MapReduce and the original Hadoop MapReduce, whose

offloading, because the amount of data to be reduced intasks are not accelerated but only run on the hosting server,

each core could be assigned as equal as possible for betterwith “Grep” and “Wordcount” applications.

data skew balancing among the offloaded reduce tasks .

aperformancecomparison

betweenH.A.

Our experimental environment information is list below.

Because of the paper length limitation, we only provide
the indexed output file image below in Figure 7

Table 1.

Host
ServerTileraBoard

Experimental Environment Specification

CPU
RAM
StorageCPU
RAMIntel E5-1660 (3.3GHz, 6-cores, HT)
128GB (DDR3-1600, 16GBx8)
Intel SSD910 (PCIe 800GB)
TILEncore-Gx36(1.2GHz,30cores)
16GB (DDR3-1333)

4.1. Evaluation of Multi-phase Merge
In this evaluation, we use the “Terasort” benchmark
scripts and dataset. By executing the TeraGen script with
Hadoop MapReduce, we generate 30GB text data first.
Then, we execute the “Terasort” benchmark script on thisdataset with two types of MapReduce strategies. The first
type is called “single-merge” MapReduce, it offloads
reduce-merge tasks to the Tilera board with the original
Hadoop’s “singe-phase merge” solution. The second is
For reduce acceleration, the offloaded reduce tasks arecalled “multi-phase merge”, which offloads reduce-merge

launched on the accelerator board node. The host process

task with our proposed “multi-phase merge” solution inof ReduceTasks reads the assigned merged file, splits themSection 3.3. The offloaded reduce-merge is defined as theinto chunks, and transfers them to the onboard process byfourth phase that described in Section 2.2. Note that bothusing DMA transfer. Note that, index information of thethe offloaded reduce-merge tasks and the offloading

merged file is used, such that items of the same key are
not split into different chunks and chunk sizes are as equal

hardware environment are totally same in both executions.
Result in Figure 9 demonstrates the significance of ouras possible. The on-board process uses multi-thread

proposed “multi-phase merge” solution. As it shows, theprocessing, and performs key-based value reduction in 30

proposed solution completes the reduce-merge tasks withparallel streams for these chunks. The intermediatenearly 2.5 times less time than original hadoop ’s “single

reduced data is transferred back to the host disk, andmerge” solution. We will further demonstrate the overall

forms one final output file at the host node. Figure 8
shows the image of our reduce acceleration flow.

MapReduce performance achievement in the next section.

775MB/s when loading the dataset from hotsing PC to
Tilera board. For the “Wordcount” applications, theoffloaded version completes in 135 seconds, which is 3.4
times faster than original MapReduce.

4.3. Efficiency Study
The results of above experiment show the feasibility ofour proposal. In this experiment, we used only one
acceleration board which implemented one CPU with has

4.2. Overall Evaluation of Proposed MapReduce
In this evaluation, we provide an overall examination of
the proposed H.A. MapReduce performance and compare
it with that of original Hadoop’s MapReduce.
Now we describe the detail of our experimental settings.
At first, the original Hadoop on host PC initializes 30 map
tasks. For the H.A. MapReduce evaluation, all these tasks
are offloaded onto Tilera board and being processed by 30
cores (1.2*30GHz) there at the same time in parallel.
While for the original Hadoop’s MapReduce evaluation,
we use the hosting P C’s resources, 12 hyper-threads(12*3.3GHz), to complete the whole map workload. Note
that, only 12 tasks are online at the same time and eachtask is designated to one thread. Our reason for this setting
is to reduce the catch switching and resource competition
overhead within a single thread if multiple tasks are being
processed there at the same time. The shuffle and merge
phases are not offloaded here, they are completed by thehosting PC in both evaluations. At last, we offload the
reduce phase in offloaded MapReduce.
We carry out above processes for three types of
evaluations. The first two are “Grep” applications; we

36 cores. We can seamlessly increase the performance byincreasing the number of boards depending on theperformance needed.
Note that the computation capability in hosting PC andthe many-core Tilera board are almost same, so that ourcompetition results should be also comparable. However,
hardware accelerated version overwhelmed the original
version. For example, although the clock speed of Tilera
core is lower, its single map task completion time (onaverage) is 3-5 times faster than that of hosting PC. This
must be due to the overhead of Java. If a C ver sion of
MapReduce is implemented, we can achieve performancecomparable to the hardware accelerated results given
above.
We also note that the merge phase of offloaded version
in “Wordcount” application takes much longer time thanoriginal version. This is because our current reduce
offloading

requires

the

MOF

(that

generatedin

reduce-merge phase) to be reduced should be located on
disk. This generates disk I/O overhead at the end of
reduce-merge phase. We will improve the merge part
implementation as one of our future work.

examine both Fuzzy and Exact match performance here. To
fairly compare the two methods, we did not use Grep
sample implementation included in the original Hadoop
distribution. We implemented a very simple Grep programusing the java standard tokenizer to split words, and regex
to check the regular expression. We used a 50GB randomly
generated dictionary, and searched for words that matched
the regular expression “a*b*c*1” or the exact string of“aabbcc1” in Fuzzy and Exact match, respectively. The
third evaluation is a “Wordcount” application. We use thesame method above to generate 10GB dictionary dataset,
and we count the number of occurrences of each word.
Note that the combiner in Map Task is enabled.
Figure 10 shows our evaluation result. For both “Grep”

5. Related work on Hardware Acceleration

applications, our proposed solution achieves almost the

There are some solutions proposed to improve the data

same execution time (81 & 75 seconds, respectively);

processing capability of single node in commodity clusters

because it has fully utilized the I/O bandwidth of our SSD

by using some specialized hardware.

Phoenix [5] implements a MapReduce model on a

Referenceshare-memory multi-core system to explore its parallelism.
However, its memory and I/O usage of one task may
severely affect others and this problem becomes criticalwhen thread number increases for processing big data.
Mars [4] accelerates MapReduce with a general GPU
platform and achieved 1.5-16X higher speed than Phoenix.
Although GPUs have an order of magnitude higher
computation power and memory bandwidth compared with
CPUs, it is not suitable for the data-intensive tasks such as
word-count, and it is hard to program due to their
special-purpose architecture design. In FPMR framework[6], dedicated processors are designed for different
applications in which map and reduce operations are doneby mappers and reducers on FPGA. The dynamic on-chipscheduling and efficient data control hide the task control,
communication, and data synchronization away fromdesigners. However, because the task scheduling and data
dispatching

are

hard

coded

on

chip,

there

is

no

reconfiguration ability in FPMR framework. [ 7] also
proposes a FPGA based acceleration framework. It studies
the scalability of this framework by increasing both FPGA
boards and servers. However, it does not support or study
the offloading of merge and sort processes. The H.A.
frameworks proposed in [8, 9] enable MR framework to
run on hybrid clusters so as to exploit the capabilities ofheterogeneous

hardware.

Theirauthors

provideascheduling policy for resource locating based on job
progress.

They

also

show

the

communication

andsynchronization overhead in data intensive applications
may still hide the benefit of hardware accelerators,
however,

they

used

commodity

network

intheirexperiments.
These previous works focus on using accelerators to
improvecommodity

systemperformance

and

the

MapReduce framework is deployed on their systems
straightforwardly only to verify the feasibility of their
proposal. No effort is paid in these works in studying or
optimization MapReduce on the H.A. infrastructure.

6. Future WorkIn our future work, we will support multiple Tilera
boardsinourframeworkforscalableandelastic

offloading solution. And we will also study the FPGA
board efficiency in improving the offloading performance,
especially for the merge tasks, in our framework. We alsofeel interested in accelerating Hive applications by using
our H.A. MapReduce proposal.

[1] Apache Hadoop (2014.2). http://hadoop.apache.org[2] J. Dean and S. Ghemawat. “MapReduce: Simplified
data processing on large clusters”, In 6th Symposium
on Operating System Design and Implementation(OSDI '04), 2004.
[3] T. Honjo and K. Oikawa, “Hardware acceleration ofHadoop MapReduce”, IEEE Intl. Conf. on Big Data
(IEEE BigData '13), 2013.
[4] B. He, W. Fang, Q. Luo, N. K. Govindaraju, and T.
Wang. “Mars: a MapReduce framework on graphics
processors”, In Proc. of the 17th intl. Conf. on
Parallel architectures and co mpilation techniques(PACT '08), 2008.
[5] C. Ranger, R. Raghuraman, A. Penmetsa, G. Bradski,
and C. Kozyrakis, “Evaluating MapReduce forMulti-core and Multiprocessor Systems ”, In Proc. of
the 13th Intel. Symposium on High Performance
Computer Architecture (HPCA '07), 2007
[6] S. Yi, B.Wang, J. Yan, Y. Wang, N. Xu, H. Yang:“FPMR: MapReduce framework on FPGA”, In Proc.
of the 18th Annual ACM/SIGDA Int l. Symposium on
Field Programmable Gate Arrays, pp. 93 –102, 2010.
[7] D. Yin, G. Li, K. Huang, “Scalable MapReduce
Framework on FPGA Accelerated Commodity
Hardware”, NEW2AN/ruSMART’12, LNCS 7469,
280-294, 2012[8] Y. Becerra, V. Beltran, D. Carrera, M. Gonzalez, J.
Torres, and E. Ayguade, “Speeding Up Distributed
MapReduce
Applications
Using
HardwareAccelerators”, in Proc. of the Intel. Conf. on Parallel
Processing (ICPP '09). 2009.
[9] J. Polo, D. Carrera, Y. Becerra, V. Beltran, J. Torres,and E. Ayguade, “Performance Management of
Accelerated MapReduce Workloads in Heterogeneous
Clusters”, in Proc. of the Intel. Conf. on Parallel
Processing (ICPP '10), 2010.
[10] F. Ahmad, S. Lee, M. Thottethodi, T.N. Vijaykumar,“PUMA: Purdue MapReduce Benchmarks Suite ”,
Purdue Technical Report TR-ECE-12-11, 2012.
[11] F. Ahmad, S. Lee, M. Thottethodi, T. N. Vijaykumar“MapReduce with communication overlap ( MaRCO)”,J. Parallel Distrib. Comput. 73, 5, 608-620, 2013.
[12] Tilera
Corporation:
TILE-Gx8036
Processor
Specification
Brief(online),availablefrom<http://www.tilera.com/sites/default/files/productbri
efs/TILEGx8036_PB033-02_web.pdf>.