日本ソフトウェア科学会第 33 回大会 (2016 年度) 講演論文集

In-memory Hadoop on supercomputers using
external memory of additional nodes
Thanh-Chung Dao 千葉 滋
In this work, we propose a variant design of in-memory Hadoop that is scalable and makes data fit into memory by using memory of other nodes. Hadoop and Spark are good choices to run MapReduce and Machinelearning algorithms, such as PageRank and K-means clustering. A supercomputer is a high-speed network
cluster consisting of thousands of compute nodes where local disk can be available (e.g. TSUBAME) or not(e.g. Fujitsu FX10). All nodes share the same disk called the central storage, and disk I/O to that storage isexpensive. To avoid disk I/O, in-memory Hadoop/Spark is required to fit into supercomputer environment.
However, memory is limited on each node and out of memory happens when data does not fit into memory.
In our proposal, we allocate a group of nodes that is only used to store data in their memory instead of thecentral storage. There is no computation tasks running on those nodes. Compared to the original Hadoop,
our experimental results show that our approach can reduce total execution time by 15% on TSUBAME
supercomputer according to naive comparison. Its performance is as good as the in-memory approach, but
it is scalable.

and reading intermediate data become a bottleneck

1 Introduction

since those data must be stored at the central stor-

This research is motivated by the challenges ofage. To solve that problem, a natural approachrunning data-intensive frameworks, such as Hadoop

is using in-memory intermediate data in order to[6] and Spark [7], on supercomputers to meet theiravoid spilling to disk. However, it is not scalable

design. A supercomputer is a big machine consist-

since memory size on each node is limited.

ing of thousands of high-performance nodes that

We propose a variant design of Hadoop that canare connected to each other through high-speed

solve the problem of scalability. It helps avoid out

network.

Not only running eﬃciently compute-of memory and makes data fit into memory. Our

intensive but also data-intensive workloads, for

proposal is motivated by a feature of supercomput-

example, clustering and classification in machine

ers that number of compute nodes are often big

learning and graph processing, are challenges on

and users can request as many nodes as they want.

supercomputers.

In aspects of productivity and

We allocate a group of nodes that is only used tomaturity, Hadoop and Spark are good choices tostore data in their memory instead of the central

run data-intensive applications on supercomputers.

storage. This design can be also applied to Spark

In today’s top supercomputers, each compute node

framework. Compared to the original Hadoop, ourdoes not have a local disk or is equipped with small

experimental results show that our approach cansize of a solid state drive (SSD). It makes writing

reduce total execution time by 15% on TSUBAME[5] supercomputer. Its performance is as good as

Thanh-Chung Dao, Shigeru Chiba, 東京大学創造情報学
専攻, Dept. of Creative Informatics, The Universityof Tokyo.

the in-memory approach, but it is scalable.

Hadoop and performance bottleneck of disk I/O
when running on supercomputers. Then, we show

Task ID

First, we show the rationale behind choosing

limitation of a natural approach that keeps data inmemory in order to avoid disk I/O. That approach

1 31 66 106 151 196 241 286 331 376

2 Motivation

Start-up (JVM & User info)
Task initializing
Data waiting
Shuffling (Data copying)
Data merging
Data reading
Task running
Task finishing

0

40

図1

2. 1 Hadoop on supercomputers
In aspects of productivity and maturity, Hadoop

80

120

Timeline (second)is called in-memory Hadoop.

Tera-sort running on TSUBAME supercomputer

using its central storage: shorter running tasks are
mappers, the longer ones are reducers.

and Spark are good choices to run data-intensive
applications on supercomputers. Both frameworks

ping phase. It is written to a node’s local disk andare used widely on commodity clusters and in in-deleted after being sent to reducers. However, in-dustry. While Hadoop is designed to be able tostead of writing to local disks, the central storage is

execute mapping and reducing functions on ter-used to store it. Figure 1 shows that time spending

abyte and petabyte-scale datasets, Spark is suit-on writing intermediate data is relatively long inable to run iterative applications, such as learningcomparison with the total execution time of map-

machine algorithms and graph processing. Hadoop

ping tasks. The figure illustrates execution timeline

provides easy-to-use interfaces that enable users toof the tera-sort application with 256 map tasks andspeed up application development rather than writ-

128 reduce tasks. Shorter running time denotes a

ing a distributed application from scratch. More-

mapper and the longer task is a reducer. In the fig-

over, Hadoop can be scaled up to thousands ofure, red part represents writing intermediate data

nodes with fault tolerance.

to a node’s local disk. Although, SSD storage can

In today’s top supercomputers, each computebe used to store intermediate data instead of using

node does not have a local disk or is equipped withthe central storage, but it is not always available

small size of a SSD. A supercomputer is often di-on supercomputers.

vided into three main components: compute nodes,
I/O nodes, and storage. It is typical that there is a

2. 2 In-memory approachcentral storage consisting of distributed disks. The

To solve the problem of bottleneck at writing andcentral storage and compute nodes are connected

reading intermediate data, a natural approach is us-

through I/O nodes and high-speed network, suching in-memory intermediate data in order to avoid

as InfiniBand. If SSD is available on each compute

spilling to disk, but memory size on each node isnode, it is only temporary storage that is used tolimited. In lieu of writing data to a node’s local

stage input data or store intermediate data. Thisdisk, a memory space is allocated and intermedi-

storage is reset each new job.

ate data is kept in this space temporarily. How big

In the Hadoop workflow, writing and reading in-size of the memory space is depends on memory

termediate data is a bottleneck on supercomput-

size on each node. On today’s top supercomputers,

ers since those data must be stored at the central

each compute node is often equipped with up to

storage. Intermediate data is outputs of the map-

256 GB memory size. However, its size is small incomparison with the central storage’s one (several
petabytes).

memory.
First, we present our own in-memory design. It is

The challenges of current in-memory design areused as a baseline for comparison with our proposal

how to scale up when size of input data increasesof external memory. Then, we show the design ofand how to avoid overhead of using next storage

external memory.

level when memory is full. M3R [4] is in-memory
Hadoop engine implemented by X10 language, but

3. 1 In-memory design

the input dataset is required to fit into memory size

In the original Hadoop design, map output isthat makes diﬃcult to scale up. Spark also supports

written to a node’s local disk (the central disk or

MapReduce pattern, and intermediate data can be

SSD on supercomputers). Map output is results af-

configured to be stored in memory or on disk. By

ter applying mapping function on datasets. Figure

default, it is kept in memory first and when there

2a shows how map output is copied from mappers

is not enough space, data is spilled to disk. Per-

to reducers. At the mapping side, map output isformance can be degraded when intermediate data

buﬀered in memory, and then when its size reaches

size is big due to overhead of writing to disk.

buﬀer size, it is spilled to the local disk. The spilled

How big memory size is allocated to the mem-

files are merged into a final map output at the end

ory space and each task process is also another re-of the mapping phase. There is a shuﬄe serversearch question. Each process running tasks should

running on each node waiting for fetching request

be allocated with big enough memory in order tofrom reducers. When a request comes, the shuf-

not slow down computation. Therefore, size of thefle server will look up map output metadata, readmemory space is subject to change when number ofagain those data from the local disk, and send backtasks running on a node increases.

to the reducer.
In the in-memory Hadoop, map output is stored

3 External memory design
In this section, we propose a variant design ofLocal Disk

Hadoop that can solve the problem presented in
the previous section. It helps avoid out of memory
and makes data fit into memory. This design can

Map Output
Mappers

ShuffleServerReducers

be also applied to Spark framework. Our proposal
is motivated by a feature of supercomputers that(a) The original Hadoop
Local Disk

number of compute nodes are often big and users
can request as many nodes as they want. We allocate a group of nodes that is only used to store

Map Output

data in their memory instead of the central stor-

Shuffle server

Mappers

Reducersage. We call those nodes memory nodes and their

In-memory storage

memory external memory. We call nodes that is

Data sendingon the same nodeInter-node communication

responsible for running applications computation
nodes. When a node in the group of computation
node does not have enough memory, it requests thegroup of memory nodes and sends data to external(b) In-memory Hadoop
図2

The original vs. In-memory Hadooptotally in memory to avoid writing to and reading(just called memory nodes). Node manager is re-from the local disk. Map output is kept in mem-sponsible for data placement and monitoring mem-ory instead of spilling to the local disk when theory left on each memory node. Since there is nobuﬀer is full. After merging, map output is sent totask running on memory nodes, size of the memory

the shuﬄe server. A memory space is created on

space can be allocated bigger.

each shuﬄe server to keep map output in memory.

In comparison with in-memory design, data send-

When a fetching request come, the shuﬄe server

ing of external memory design is more complex.

looks up the memory space and is able to send backFirst, the mapper requests its shuﬄe server to send

immediately to the reducer.

map output. If the shuﬄe server still has empty
space, data is sent then. If the shuﬄe server’s mem-

3. 2 External memory design

ory space is full, the mapper requests the memory

Size of the memory space on each shuﬄe server

node manager to send map output. The node man-is limited, so out of memory might happen during

ager will find a relevant place and send back to thedata fetching. For example, the memory space size

mapper. Then, the mapper starts sending map out-

is 8 GB and there are 20 map tasks running on the

put to the specified memory node. When finishingnode. If each input data is 512 MB and runningsending map output, the mapper also informs thetera-sort application, the total size of map output

shuﬄe server about the location of map output on

is 10 GB. Those data cannot fit into the memory

memory nodes.

space. Although map output can be deleted rightDue to waste of CPU time on memory nodes,

after it is sent to the corresponding reducer, thata question is that should we run tasks on thoseout of memory error can happen at any time.

nodes? Since task process requires a certain mem-

Our approach is to create a group of nodes (called

ory to do computation, size of external memory

memory nodes) used only for storing map output

must be decreased. Moreover, disk I/O and net-

and there is no map or reduce task running on thosework communication also happen when running

nodes. When the memory space of a shuﬄe servertasks that may make memory nodes busier.

is full, map output is sent to memory nodes. In
Figure 3, the group of memory nodes consist of one
memory node manager (master) and slave nodes

4 Experimental evaluation
In this section, we evaluate whether external
memory usage adds any overhead to total execution

Shuffle server
Mapperstime. We also show how many memory nodes are
Reducersenough when running tera-sort application. Our
Map Output

Memory node

Memory node

Memory node

Memory nodeexperiments are conducted on TSUBAME supercomputer where each compute node is equipped

In-memory storage…

Request exchangeInter-node communication

図3

with Intel Xeon X5670 2.93GHz and 54GB memory. We are still working on experiments in orderto improve their accuracy, so the follows are theMemory nodeManager

External memory Hadoop

preliminary results and the comparison is naive.

nal memory, we decrease in-memory space on each

1 35 75 120 171 222 273 324 375

Task ID

shown in Figure 4b. In the experiment of exter-

computation nodes in order to make data be moved
to memory nodes. Figure 4c shows that reducers
Start-up (JVM & User info)
Task initializing
Data waiting
Shuffling (Data copying)
Data merging
Data reading
Task running
Task finishing

0

40

80

120 160 200 240 280 320 360 400 440 480

also complete at 230 seconds.

4. 2 How many memory nodes are enoughSuppose we have a fixed number of computation

Timeline (second)(a) The original Hadoop

nodes (denoted by Ncomp = Constant). Nmem de-Task ID

1 35 75 121 172 223 274 325 376

notes number of memory nodes (that used only forstoring data). Exec denotes execution time when
number of memory nodes is Nmem . We need to findStart-up (JVM & User info)
Task initializing
Data waiting
Shuffling (Data copying)
Data merging
Data reading
Task running
Task finishing

0

40

80

120 160 200 240 280 320 360 400 440 480

a relevant Nmem where Exec is fastest:
M in(Exec(Ncomp , Nmem ))In our current design, number of memory nodes
is estimated roughly based on the size of input data.

Timeline (second)

Our assumption is that total size of map output isthat total size of memory space on shuﬄe servers
and memory nodes must be bigger than size of inStart-up (JVM & User info)
Task initializing
Data waiting
Shuffling (Data copying)
Data merging
Data reading
Task running
Task finishing

0

40

80

put data. It might not achieve the best performance of data fetching, but it helps prevent outof memory error. That error is more serious if it

120 160 200 240 280 320 360 400 440 480
Timeline (second)(c) External memory Hadoop
図4equal to one of input data. Therefore, number ofmemory node is chosen that satisfies a condition

1 35 75 121 172 223 274 325 376

Task ID(b) In-memory Hadoop

External memory performance (Tera-sorton TSUBAME)

happens since the whole task will be restarted.
We have conducted an experiment by keeping thesame number of computation nodes and changing
number of memory nodes. The purpose of this ex-

4. 1 External memory performance

approach but faster than the original Hadoop. We
compare three test cases: the original Hadoop, in-

3000
2000

of using external memory is as good as in-memory

Time (Milliseconds)

This experiment is aimed to show performanceory Hadoop). Figure 4a shows that reducers finish

1000

memory Hadoop, and our approach (external memData copying time4

after 270 seconds in the original Hadoop. A bar

6

8

10

12

14

16

Number of memory nodes

with longer execution time denotes a reducer. It
is not surprising that in-memory Hadoop, it takes
only 230 seconds to finish the reducing phase as

図5

How many memory nodes are enoughperiment is to check whether our current approachof finding number of memory nodes is eﬀective or

6 Conclusion

not. We run tera-sort application on 32 computa-

In this work, we propose a variant design of in-tion nodes. Number of memory nodes ranges frommemory Hadoop that is scalable and makes data fit

4 to 16. The size of input data is 64 GB generated

into memory by using memory of other nodes. In

using tera-gen application. According to our esti-

our proposal, we allocate a group of nodes that ismation, required number of memory nodes should

only used to store data in their memory instead ofbe 4 ∼ 8 nodes. We measure data copying time (be-the central storage. There is no computation tasks

tween shuﬄe servers and memory nodes and reduc-running on those nodes. Compared to the origi-ers) rather than total execution time since memory

nal Hadoop, our experimental results show that ournodes aﬀect mainly communication time. Figure

approach can reduce total execution time by 15%

5 shows that when number of memory nodes is 8,

on TSUBAME supercomputer according to naive

data is copied fastest. However, if increasing num-comparison. Its performance is as good as the in-ber of memory nodes from 8 to 16, copying time ismemory approach, but it is scalable. We are stillslower.

working on experiments in order to improve their

5 Related work

accuracy. As future work, we discuss several tech-

Memcached is a distributed caching system us-

nical issues that we are still working in progress as

ing key-value pair storage. It is widely used tocache database queries. Our external memory de-follows:
Communication among processes.

Since

sign can be considered as combination of Hadoop

data is kept in memory and sent/received amongand Memcached. To the best our knowledge, there

processes, a fast data exchange protocol is required

is no study carried out to enable Hadoop MapRe-in order to reduce overhead of network communi-

duce workflow to store intermediate data implic-

cation. On supercomputer environment, Message

itly on Memcached system. We also evaluate somePassing Interface (MPI) is the de facto commu-

aspects of that combination. RDMA-based Mem-nication protocol and often optimized to achievecached [3] is an extension that speeds up inter-nodehigh throughput and low latency. In the originalcommunication by using RDMA. In our current de-Hadoop, the shuﬄe server is based on TCP protocol

sign, we use MPI communication to exchange data

and reducers make HTTP requests to the server. In

among processes on diﬀerent nodes.

our current design, data sending from mappers to

HaLoop [1] is a variant of Hadoop that supports

a shuﬄe server and data fetching between reducers

caching input, intermediate, and output data in or-

and a shuﬄe server is implemented using MPI com-der to reuse data for running iterative applications.

munication. The shuﬄe server uses non-blocking

Intermediate data is written to a node’s local disk

(asynchronized) sending/receiving, so it can han-using a hash algorithm that helps speed up read-dle multiple requests from reducers. To keep MPI

ing later. On supercomputers, however, instead ofconnection between processes, we use HPC-Reuse

local disk, writing and reading to/from the central[2] framework.

storage is a bottleneck.

Data placement on memory nodes. What isthe best strategy of storing map output in external
memory? Should we distribute map output to as
many memory nodes as possible or just store on a

memory node and when it is full, the next memory
node will be used? The former helps increase fetching throughput since reducers can request to morememory nodes. The latter is useful if map output ispackaged and sent once to a reducer. In the current
design, we have just implemented a round robinscheduler on the memory node manager. Data will
be distributed on all memory nodes.
参 考 文 献[1] Bu, Y., Howe, B., Balazinska, M., and Ernst,
M. D.: HaLoop: eﬃcient iterative data processing
on large clusters, Proceedings of the VLDB Endowment, Vol. 3, No. 1-2(2010), pp. 285–296.
[2] Dao, T.-C. and Chiba, S.: HPC-Reuse: Efficient Process Creation for Running MPI and
Hadoop MapReduce on Supercomputers, 2016 16th

IEEE/ACM International Symposium on Cluster,
Cloud and Grid Computing (CCGrid), IEEE, 2016,
pp. 342–345.
[3] Jose, J., Subramoni, H., Luo, M., Zhang, M.,
Huang, J., Wasi-ur Rahman, M., Islam, N. S.,
Ouyang, X., Wang, H., Sur, S., et al.: Memcacheddesign on high performance rdma capable interconnects, 2011 International Conference on Parallel
Processing, IEEE, 2011, pp. 743–752.
[4] Shinnar, A., Cunningham, D., Saraswat, V., andHerta, B.: M3R: increased performance for inmemory Hadoop jobs, Proceedings of the VLDB Endowment, Vol. 5, No. 12(2012), pp. 1736–1747.
[5] TSUBAME: http://tsubame.gsic.titech.ac.jp/en/top,
2016.
[6] White, T.: Hadoop: The definitive guide, ”
O’Reilly Media, Inc.”, 2012.
[7] Zaharia, M., Chowdhury, M., Franklin, M. J.,
Shenker, S., and Stoica, I.: Spark: cluster computing with working sets, Proceedings of the 2nd
USENIX conference on Hot topics in cloud computing, 2010, pp. 10–10.