MPI and MapReduce

(1)

MPI and MapReduce

CCGSC 2010 Flat Rock NC

September 8 2010

Geoffrey Fox

[email protected]

http://www.infomall.org http://www.futuregrid.org

Director, Digital Science Center, Pervasive Technology Institute

(2)

MapReduce

• _{Implementations (Hadoop – Java; Dryad – Windows) support:}

– _{Splitting of data with customized file systems}

– _{Passing the output of map functions to reduce functions}

– _{Sorting the inputs to the reduce function based on the intermediate keys} – _{Quality of service}

• _{20 petabytes per day (on an average of 400 machines) processed by}

Google using MapReduce September 2007

Map(Key, Value)

Reduce(Key, List<Value>)

Data Partitions

Reduce Outputs

(3)

MapReduce “File/Data Repository” Parallelism

Instruments

Disks Map₁ Map₂ Map₃ Reduce

Communication

Map = (data parallel) computation reading and writing data Reduce = Collective/Consolidation phase e.g. forming multiple global sums as in histogram

Portals /Users

MPI or Iterative MapReduce

Map Reduce Map Reduce Map

(4)

Typical Application Challenge:

DNA Sequencing Pipeline

Illumina/Solexa Roche/454 Life Sciences Applied Biosystems/SOLiD

Modern Commercial Gene Sequencers

Internet Read Alignment Read Alignment Visualization Visualization Blocking

Blocking _Alignment/Sequence

Assembly Sequence Alignment/ Assembly MDS MDS Dissimilarity Matrix N(N-1)/2 values Dissimilarity Matrix N(N-1)/2 values FASTA File N Sequences FASTA File N Sequences block Pairings Pairwise clustering Pairwise clustering MapReduce MPI

(5)

Metagenomics

This visualizes results of dimension reduction to 3D of 30000 gene sequences from an environmental sample. The many different genes are classified by clustering algorithm and visualized by MDS

(6)

All-Pairs Using MPI or DryadLINQ

35339 50000 0

2000 4000 6000 8000 10000 12000 14000 16000 18000 20000

DryadLINQ MPI

Calculate Pairwise Distances (Smith Waterman Gotoh)

125 million distances 4 hours & 46 minutes 125 million distances 4 hours & 46 minutes

• Calculate pairwise distances for a collection of genes (used for clustering, MDS)

• _{Fine grained tasks in MPI}

• Coarse grained tasks in DryadLINQ

• Performed on 768 cores (Tempest Cluster)

(7)

Smith Waterman

MPI DryadLINQ Hadoop

10000 20000 30000 40000

0.000 0.005 0.010 0.015 0.020 0.025 Hadoop SW-G MPI SW-G DryadLINQ SW-G

No. of Sequences

Ti m e p e r A ct u al C al cu la ti o n ( m s)

(8)

Twister(MapReduce++)

• Streaming based communication

• Intermediate results are directly transferred from the map tasks to the reduce tasks –

eliminates local files

• Cacheablemap/reduce tasks

•Static data remains in memory

• Combine phase to combine reductions

• User Program is the composer of MapReduce computations

• Extendsthe MapReduce model to iterative

computations Data Split D _MR Driver User Program

Pub/Sub Broker Network

D File System M R M R M R M R Worker Nodes M R D Map Worker Reduce Worker MRDeamon Data Read/Write Communication

Reduce (Key, List<Value>) Reduce (Key, List<Value>)

Iterate

Map(Key, Value) Map(Key, Value)

Combine (Key, List<Value>) Combine (Key, List<Value>) User Program User Program Close() Close() Configure() Configure() Static data Static data δ flow δ flow

(9)

Iterative and non-Iterative Computations

K-means

Performance of K-Means

(10)

Matrix Multiplication 64 cores

Square blocks Twister

Row/Col decomp Twister

(11)

Overhead OpenMPI v Twister

negative overhead due to cache

(12)

Performance of Pagerank using

ClueWeb Data (Time for 20 iterations)

(13)

Fault Tolerance and MapReduce

• _MPI

_{does “maps” followed by “communication” including}

“reduce” but does this iteratively

• _{There must (for most communication patterns of interest) be a}

strict synchronization

at end of each communication phase

–

_{Thus if a}

_{process fails then everything grinds to a halt}

• _{In MapReduce, all Map processes and all reduce processes are}

independent

and stateless and read and write to disks

–

_{As 1 or 2 (reduce+map) iterations, no difficult synchronization issues}

• _Thus

_{failures can easily be recovered}

_{by rerunning process without}

other jobs hanging around waiting

• _{Re-examine MPI fault tolerance in light of MapReduce}

–

_{Relevant for Exascale?}

(14)

MPI & Iterative MapReduce papers

• _{MapReduce on MPI}_{Torsten Hoefler, Andrew Lumsdaine and Jack Dongarra,}_{Towards Efficient}

MapReduce Using MPI, Recent Advances in Parallel Virtual Machine and Message Passing Interface Lecture Notes in Computer Science, 2009, Volume 5759/2009, 240-249

• MPI with generalized MapReduce

• Jaliya Ekanayake, Hui Li, Bingjing Zhang, Thilina Gunarathne, Seung-Hee Bae, Judy Qiu, Geoffrey Fox Twister: A Runtime for Iterative MapReduce, Proceedings of the First International Workshop on MapReduce and its Applications of ACM HPDC 2010 conference, Chicago, Illinois, June 20-25, 2010 http://grids.ucs.indiana.edu/ptliupages/publications/twister__hpdc_mapreduce.pdf

http://www.iterativemapreduce.org/

• _{Grzegorz Malewicz, Matthew H. Austern, Aart J. C. Bik, James C. Dehnert, Ilan Horn, Naty Leiser,}

and Grzegorz Czajkowski Pregel: A System for Large-Scale Graph Processing, Proceedings of the 2010 international conference on Management of data Indianapolis, Indiana, USA Pages: 135-146 2010

• Yingyi Bu, Bill Howe, Magdalena Balazinska, Michael D. Ernst HaLoop: Efficient Iterative Data Processing on Large Clusters, Proceedings of the VLDB Endowment, Vol. 3, No. 1, The 36th International Conference on Very Large Data Bases, September 1317, 2010, Singapore.

• _{Matei Zaharia, Mosharaf Chowdhury, Michael J. Franklin, Scott Shenker, Ion Stoica}_{Spark: Cluster}

Computing with Working Sets poster at

(15)

(16)

Scaled Timing with

Azure/Amazon MapReduce

64 * 1024

96 * 1536

128 * 2048

160 * 2560

192 * 3072 1000 1100 1200 1300 1400 1500 1600 1700 1800 1900

Cap3 Sequence Assembly

Azure MapReduce Amazon EMR

Hadoop Bare Metal Hadoop on EC2

Number of Cores * Number of files

T

im

e

(s

(17)

Cap3 Cost

0 2 4 6 8 10 12 14 16 18

Azure MapReduce Amazon EMR

Hadoop on EC2

Num. Cores * Num. Files

C

o

st

(

(18)

Smith Waterman: “Scaled Speedup” Timing

0 500 1000 1500 2000 2500 3000

Azure MR

Amazon EMR

Hadoop on EC2

Hadoop on Bare Metal

Num. of Cores * Num. of Blocks

A

d

ju

st

e

d

T

im

e

(

(19)

Smith Waterman: daily effect

W ed_D

ay W

ed_N igh_t Th u D ay Th u N igh_t

Fri_D

ay Fri N_igh t Sa t D ay Sa t N igh_t Su n D ay Su n N igh_t M on D ay M on_N

igh_t Tu e D ay Tu e N igh_t 1000 1100 1200 1300 1400 1500 1600 1700 EMR Azure MR

Azure MR Adj.

Ti

m

e

(

(20)

SWG Cost

64 * 1024 96 * 1536 128 * 2048 160 * 2560 192 * 3072 0

5 10 15 20 25 30

AzureMR Amazon EMR Hadoop on EC2

Num. Cores * Num. Blocks

C

o

st

(

(21)

TwisterMPIReduce

• _{Runtime package supporting subset of MPI}

mapped to Twister

• _{Set-up, Barrier, Broadcast, Reduce}

TwisterMPIReduce TwisterMPIReduce PairwiseClustering MPI PairwiseClustering MPI Multi Dimensional Scaling MPI Multi Dimensional Scaling MPI Generative Topographic Mapping MPI Generative

Topographic Mapping MPI

Other … Other …

Azure Twister (C# C++)

Azure Twister (C# C++) _{Java Twister} Java Twister

Microsoft Azure

(22)

Some Issues with AzureTwister and

AzureMapReduce

• _{Transporting data to Azure}

_{: Blobs (HTTP), Drives (GridFTP}

etc.), Fedex disks

• _{Intermediate data Transfer}

_{: Blobs (current choice) versus}

Drives (should be faster but don’t seem to be)

• _{Azure Table v Azure SQL}

_{: Handle all metadata}

• _{Messaging Queues}

_{: Use real publish-subscribe system in}

place of Azure Queues to get scaling (?) with multiple brokers

– especially AzureTwister

• _{Azure Affinity Groups}

_{: Could allow better data-compute and}

(23)

Research Issues

• _{Clouds are suitable for “Loosely coupled” data parallel applications}

• _{“Map Only” (really pleasingly parallel) certainly run well on clouds}

(subject to data affinity) with many programming paradigms

• _{Parallel FFT and adaptive mesh PDE solver very bad on MapReduce}

but suitable for classic MPI engines.

• _{MapReduce is more dynamic and fault tolerant than MPI; it is}

simpler and easier to use

• _{Is there an intermediate class of problems for which Iterative}

MapReduce useful?

– Long running processes?

– Mutable data small in size compared to fixed data(base)?

– Only support reductions?

– Is it really different from a fault tolerant MPI?

– Multicore implementation

– Link to HDFS or equivalent data parallel file system

(24)

FutureGrid in a Nutshell

• _{FutureGrid provides a testbed with a wide variety of computing services}

to its users

– _{Supporting users developing new applications and new middleware using}_Cloud,

Grid and Parallel computing (Hypervisors – Xen, KVM, ScaleMP, Linux, Windows, Nimbus, Eucalyptus, Hadoop, Globus, Unicore, MPI, OpenMP …)

– _{Software supported by FutureGrid or users}

– _{~5000 dedicated cores distributed across country}

• _{The FutureGrid testbed provides to its users:}

– _{A rich development and testing platform for middleware and application users}

looking at interoperability, functionality and performance

– _{A rich}_{education and teaching}_{platform for advanced cyberinfrastructure classes}

• _{Each use of FutureGrid is an}

_experiment

_{that is}

_reproducible

• _Cloud

_{infrastructure supports loading of general images on}

_Hypervisors

(25)

FutureGrid: a Grid/Cloud

Testbed

• Operational: IU Cray operational; IU , UCSD, UF & UC IBM iDataPlex operational

• _Network_,_NID_operational

• _TACC_{Dell running acceptance tests – ready ~September 15}

NID: Network

Impairment Device

Private

(26)

FutureGrid Dynamic

Provisioning Results

4 8 16 32

0:00:00 0:00:43 0:01:26 0:02:09 0:02:52 0:03:36 0:04:19

Total Provisioning Time minutes

Time

Time elapsed between requesting a job and the jobs reported start time on the provisioned node. The numbers here are an average of 2 sets of experiments.

Time minutes

(27)