Advances in Clouds and their application to Data Intensive problems

(1)

https://portal.futuregrid.org

Advances in Clouds and their

application to Data Intensive

problems

Electrical Engineering

University of Southern California

February 24 2012

Geoffrey Fox

[email protected]

http://www.infomall.org

http://www.salsahpc.org

Director, Digital Science Center, Pervasive Technology Institute

Associate Dean for Research and Graduate Studies, School of Informatics and Computing Indiana University Bloomington

(2)

Topics Covered

• _{Broad Overview: Data Deluge to Clouds}

• _{Internet of Things: Sensor Grids supported as pleasingly}

parallel applications on clouds

• _{MapReduce and Iterative MapReduce for non trivial parallel}

“analytics” on Clouds

• _{MapReduce and Twister on Azure}

• _{Clouds Grids and Supercomputers: Infrastructure and}

Applications

• _FutureGrid

• _{Abstract Image management on FutureGrid}

(3)

Broad Overview:

Data Deluge to Clouds

(4)

Some Trends

The Data Deluge

is clear trend from Commercial (Amazon,

e-commerce) , Community (Facebook, Search) and Scientific

applications

Light weight clients

from smartphones, tablets to sensors

Multicore

reawakening parallel computing

Exascale initiatives

will continue drive to high end with a

simulation orientation

Clouds

with cheaper, greener, easier to use

IT for (some)

applications

New jobs

associated with new curricula

Clouds as a distributed system

(classic CS courses)

Data Analytics

(Important theme at SC11)

Network/Web Science

(5)

Some Data sizes

~40 10

9

_{Web pages}

_{at ~300 kilobytes each = 10 Petabytes}

Youtube

48 hours video uploaded per minute;

in 2 months in 2010, uploaded more than total NBC ABC CBS

~2.5 petabytes per year uploaded?

LHC

15 petabytes per year

Radiology

69 petabytes per year

Square Kilometer Array Telescope

will be 100

terabits/second

Earth Observation

becoming ~4 petabytes per year

Earthquake Science

– few terabytes

total

today

PolarGrid

– 100’s terabytes/year

Exascale simulation

data dumps – terabytes/second

(6)

Why need cost effective

Computing!

(7)

Clouds Offer

From different points of view

• _{Features from NIST:}

–

_{On-demand service (elastic);}

–

_{Broad network access;}

–

_{Resource pooling;}

–

_{Flexible resource allocation;}

–

_{Measured service}

• _{Economies of scale}

_{in performance and electrical power}

_{(Green IT)}

• _{Powerful new}

_{software models}

–

_{Platform as a Service}

_{is not an alternative to}

_{Infrastructure as a}

Service –

it is instead an incredible valued added

–

_{Amazon is as much PaaS as Azure}

(8)

The Google gmail example

• _{http://www.google.com/green/pdfs/google-green-computing.pdf}

• _{Clouds win by efficient resource use and efficient data centers}

8

Business

Type Number of users # servers IT Power per user PUE (Power Usage effectiveness)

Total Power per

user

Annual Energy per

user

Small 50 2 8W 2.5 20W 175 kWh

Medium 500 2 1.8W 1.8 3.2W 28.4 kWh

Large 10000 12 0.54W 1.6 0.9W 7.6 kWh

Gmail

(9)

Gartner 2009 Hype Curve

Clouds, Web2.0, Green IT

(10)

(11)

Jobs v. Countries

(12)

2 Aspects of Cloud Computing:

Infrastructure and Runtimes

• _{Cloud infrastructure:}

_{outsourcing of servers, computing, data, file}

space, utility computing, etc..

• _{Cloud runtimes or Platform:}

_{tools to do data-parallel (and other)}

computations. Valid on Clouds and traditional clusters

–

_{Apache Hadoop, Google}

_MapReduce

_{, Microsoft Dryad, Bigtable,}

Chubby and others

–

_{MapReduce designed for information retrieval but is excellent for}

a wide range of

science data analysis applications

–

_{Can also do much traditional parallel computing for data-mining}

if extended to support

iterative

operations

(13)

Internet of Things: Sensor Grids

supported as pleasingly parallel

applications on clouds

(14)

Internet of Things: Sensor Grids

A pleasingly parallel example on Clouds

A

sensor

(“

Thing

”) is any source or sink of time series

In the thin client era, smart phones, Kindles, tablets, Kinects, web-cams are

sensors

Robots, distributed instruments such as environmental measures are sensors

Web pages, Googledocs, Office 365, WebEx are sensors

Ubiquitous Cities/Homes are full of sensors

They have IP address on Internet

Sensors – being intrinsically distributed are

Grids

However natural implementation uses

clouds

to consolidate and

control and collaborate with sensors

Sensors are typically “small” and have

pleasingly parallel cloud

implementations

(15)

Sensors as a Service

Sensors as a Service

Sensor

Processing as

a Service

(MapReduce)

A larger sensor ………

(16)

More on Sensors

• _{Hardware sensors}

• 1. GPS device

• 2. RFID reader and tags’ signal strength

• _{3. Lego NXT Robots with:}

• a. Light b. Sound

• c. Touch d. Ultrasonic

• e. Compass f. Gyro

• g. Accelerometer h. Temperature

• _{4. Wii Remote Controller} • 5. Android Phones and Tablets

• 6. IP Cameras/microphones (RTSP, RTMP,HTTP)

• _{7. Web Cameras/microphones} •

• _{Computational services (software sensors)} • 1. Video Edge Detection

• 2. Video Face Detection

• _{3. Twitter Sensor}

• 4. Collaborative Sensors

• _{a. Chat (with language translation)} • _{b. File Transfer}

Future Work

HexaCopter from jDrones

(17)

(18)

Performance of Pub-Sub Cloud Brokers

• _{High end sensors equivalent to Kinect or MPEG4 TRENDnet}

TV-IP422WN camera at about 1.8Mbps per sensor instance

• _{OpenStack hosted sensors and middleware}

18

0 50 100 150 200 250 300 350 1

10 100 1000

Jitter versus Packet Number (Time)

10 Clients 50 Clients 100 Clients 200 Clients Packet Number Ji tt e r in m s

0 50 100 150 200 250 300 0 2 4 6 8 10 12

Single Broker Average Message Latency

Number of Clients

(19)

MapReduce and Iterative

MapReduce for non trivial

parallel applications on Clouds

(20)

MapReduce “File/Data Repository” Parallelism

Instruments

Disks Map1 Map2 Map3

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

(21)

Twister v0.9

March 15, 2011

New Interfaces for Iterative MapReduce Programming

http://www.iterativemapreduce.org/

SALSA Group

Bingjing Zhang, Yang Ruan, Tak-Lon Wu, Judy Qiu, Adam

Hughes, Geoffrey Fox,

Applying Twister to Scientific

Applications

, Proceedings of IEEE CloudCom 2010

Conference, Indianapolis, November 30-December 3, 2010

Twister4Azure

released May 2011

http://salsahpc.indiana.edu/twister4azure/

MapReduceRoles4Azure

available for some time at

(22)

K-Means Clustering

• _{Iteratively refining operation}

• _{Typical MapReduce runtimes incur extremely high overheads}

– _{New maps/reducers/vertices in every iteration} – _{File system based communication}

• _{Long running tasks and faste}

_{r communication in Twister enables it to}

perform close to MPI

Time for 20 iterations

map map

reduce

Compute the distance to each data point from each cluster center and assign points to cluster centers Compute new cluster centers

Compute new cluster centers

(23)

Twister

• 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

(24)

SWG Sequence Alignment Performance

(25)

Performance of Pagerank using

ClueWeb Data (Time for 20 iterations)

(26)

Map Collective Model (Judy Qiu)

• _{Combine MPI and MapReduce ideas}

• _{Implement collectives optimally on Infiniband,}

Azure, Amazon ……

26

Input

map

Generalized Reduce Initial Collective Step Network of Brokers

Final Collective Step Network of Brokers

Iterate

Compute

Communicate

Compute

Communicate

Most Parallel Programs consist of loosely synchronized succession of compute-communicate stages.

MPI Collectives supported this

(27)

Execution Time Improvements

Circle Fouettes (Direct Download) Fouettes (MST Gather) 0.00 2000.00 4000.00 6000.00 8000.00 10000.00 12000.00 14000.00 12675.41 3054.91 3190.17

Kmeans, 600 MB centroids (150000 500D points), 640 data points, 80 nodes, 2 switches, MST Broadcasting, 50 iterations

Circle Fouettes (Direct Download) Fouettes (MST Gather)

To ta l E xe cu ti o n T im e ( U n it : Se co n d s)

(28)

Twister on Azure

(29)

High Level Flow Twister4Azure



_{Merge Step}



_{In-Memory Caching of static data}



_{Cache aware scheduling}



_Azure

_Queues

_{for scheduling,}

_Tables

_{to store meta-data and}

monitoring data,

Blobs

for input/output/intermediate data

(30)

Simple Performance Comparisons

50% 55% 60% 65% 70% 75% 80% 85% 90% 95% 100% P a ra lle l E ffi ci en cy

Num. of Cores * Num. of Files

Twister4Azure Amazon EMR Apache Hadoop

Cap3 Sequence Assembly

(31)

Look at one problem in detail

• _{Visualizing Metagenomics where sequences are ~1000}

dimensions

• _{Map sequences to 3D so you can visualize}

• _{Minimize Stress}

• _{Improve with deterministic annealing (gives lower stress}

with less variation between random starts)

• _{Need to}

_iterate

_{Expectation Maximization}

• _N

2

_{dissimilarities (Smith Waterman, Needleman-Wunsch,}

Blast)



_{i j}

• _{Communicate N positions}

_X

_{between steps}

(32)

(33)

440K Interpolated

(34)

Multi-Dimensional-Scaling

• _{Many iterations}

• _{Memory & Data intensive}

• _{3 Map Reduce jobs per iteration}

• X

_k

= invV * B(X

_(k-1)

) * X

_(k-1)

• _{2 matrix vector multiplications termed BC and X}

BC: Calculate BX

Map

Map ReduceReduce MergeMerge

X: Calculate invV (BX)

Map

Calculate Stress

Map

(35)

Performance – Multi Dimensional

Scaling

Weak Scaling Data Size Scaling

Performance adjusted for sequential performance difference

(36)

Performance – Kmeans

Clustering

Number of Executing Map Task Histogram

Strong Scaling with 128M Data Points

Weak Scaling Task Execution Time Histogram

First iteration performs the initial data fetch

Overhead between iterations Overhead between iterations

Scales better than Hadoop on bare metal

(37)

Data Caching

• _{In-memory and disk caching}

–

_{Loop-invariant data}

–

_{Any other shared data (eg: broadcast data)}

• _{Loop-invariant and other shared data (eg: broadcast}

data)

• _{Disk Caching – up to 50% speedups over non-cached}

• _{In-Memory – up to 22% speedups over disk caching}

(38)

Memory Caching performance anomaly

• _{In-Memory caching}

– _{Inconsistencies with high memory}

applications

• Disk caching

– _{performance inconsistencies with disk I/O}

• _{.Net Memory Mapped file based}

caching

– _{Stable performance}

– _{Works better on larger instances}

Mechanism Instance Type

Total Execution Time (s)

Task Time (BCCalc) Map Fn_{Time (BCCalc)}

Average

(ms) STDEV (ms) # of slow tasks Average (ms) STDEV (ms) # of slow tasks

Disk Cache only small * 1 2676 6,390 750 40 3,662 131 0

In-Memory

Cache small * 1 2072 4,052 895 140 3,924 877 143

Memory Mapped

(39)

Iterative MapReduce Collective

Communication Operations

• _{Supports common higher-level communication patterns}

–

_{Substitutes certain steps of the computation}

• _{Framework can optimize these operations transparently to}

users

• _{Ease of use}

–

_{Don’t have to implement the steps substituted by these}

operations

• _SumReduce

–

_{Addition of single value numerical outputs of Map Tasks}

• _AllGather

(40)

Clouds Grids and

Supercomputers: Infrastructure

and Applications

(41)

What Applications work in Clouds

• _Workflow

_and

_Services

• _{Pleasingly parallel}

_{applications of all sorts analyzing}

roughly independent data or spawning independent

simulations including

–

_{Long tail}

_{of science}

–

_{Integration of distributed}

_sensor

_data

• _{Science Gateways}

_{and portals}

• _{Commercial and Science Data analytics}

_{that can use}

MapReduce

(

some of such apps) or its

iterative

variants

(most analytic apps)

• _{Note Data Analysis requirements not well articulated in}

many fields – See http://www.delsall.org for life sciences

(42)

Clouds and Grids/HPC

• _{Synchronization/communication Performance}

Grids > Clouds > HPC Systems

• _{Clouds appear to execute effectively Grid workloads but}

are not easily used for closely coupled HPC applications

• _{Service Oriented Architectures}

_and

_workflow

_{appear to}

work similarly in both grids and clouds

• _{Assume for immediate future, science supported by a}

mixture of

–

_{Clouds – data analysis (and pleasingly parallel)}

–

_{Grids/High Throughput Systems (moving to clouds as}

convenient)

(43)

43

(a) Map Only

(a) Map Only (c) Iterative _Synchronous_Synchronous(d) Loosely (d) Loosely MapReduce (c) Iterative MapReduce (b) Classic MapReduce (b) Classic MapReduce Input Input map map reducereduce Input Input map map reduce reduce Iterations Iterations Input Input Output Output map map P_ij P_ij BLAST Analysis Smith-Waterman Distances Parametric sweeps PolarGrid Matlab data analysis

BLAST Analysis Smith-Waterman Distances

Parametric sweeps PolarGrid Matlab data analysis

High Energy Physics (HEP) Histograms Distributed search Distributed sorting Information retrieval High Energy Physics (HEP) Histograms Distributed search Distributed sorting Information retrieval

Many MPI scientific applications such as solving differential equations and particle dynamics Many MPI scientific applications such as solving differential equations and particle dynamics

Domain of MapReduce and Iterative Extensions

Domain of MapReduce and Iterative Extensions MPIMPI

Expectation maximization clustering e.g. Kmeans Linear Algebra

Multimensional Scaling Page Rank

(44)

FutureGrid in a Nutshell

(45)

FutureGrid key Concepts I

• _{FutureGrid is an}

_{international testbed}

_{modeled on Grid5000}

• _{Supporting international}

_{Computer Science}

_and

_{Computational}

Science

research in cloud, grid and parallel computing (HPC)

–

_{Industry and Academia}

–

_{Note much of current use Education, Computer Science Systems}

and Biology/Bioinformatics

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

–

_{A flexible development and testing platform for middleware}

and application users looking at

interoperability

,

functionality

,

performance

or

evaluation

–

_{Each use of FutureGrid is an}

_experiment

_{that is}

_reproducible

–

_{A rich}

_{education and teaching}

_{platform for advanced}

(46)

FutureGrid key Concepts II

• Rather than loading images onto VM’s, FutureGrid supports

Cloud, Grid and Parallel computing

environments by

dynamically provisioning

software as needed onto “bare-metal”

using Moab/xCAT

– Image library

for MPI, OpenMP, Hadoop, Dryad, gLite, Unicore, Globus,

Xen, ScaleMP (distributed Shared Memory),

Nimbus

,

Eucalyptus

,

OpenStack

, KVM, Windows …..

• Growth comes from users depositing novel images in library

• FutureGrid has ~4000 (will grow to ~5000) distributed cores

with a dedicated network and a Spirent XGEM network fault

and delay generator

Image1

Image1 _Image2Image2

…

_ImageNImageN

Load

(47)

Cores

11TF IU 1024 IBM

4TF IU 192 12 TB Disk 192 GB mem, GPU on 8 nodes 6TF IU 672 Cray XT5M

8TF TACC 768 Dell 7TF SDSC 672 IBM 2TF Florida 256 IBM 7TF Chicago 672 IBM

FutureGrid:

a Grid/Cloud/HPC Testbed

Private

Public FG Network

NID: Network Impairment Device

Upgrades include

Larger memory 192GB/node 12 TB disk per node

(48)

FutureGrid Partners

• _{Indiana University}

_{(Architecture, core software, Support)}

• _{Purdue University}

_{(HTC Hardware)}

• _{San Diego Supercomputer Center}

_{at University of California San Diego}

(INCA, Monitoring)

• _{University of Chicago}

_{/Argonne National Labs (Nimbus)}

• _{University of Florida}

_{(ViNE, Education and Outreach)}

• _{University of Southern California Information Sciences (Pegasus to manage}

experiments)

• _{University of Tennessee Knoxville (Benchmarking)}

• _{University of Texas at Austin}

_{/Texas Advanced Computing Center (Portal)}

• _{University of Virginia (OGF, Advisory Board and allocation)}

• _{Center for Information Services and GWT-TUD from Technische Universtität}

Dresden. (VAMPIR)

(49)

5 Use Types for FutureGrid

• _~122

_{approved projects over last 12 months}

• _{Training Education and Outreach}

_(11%)

–

_{Semester and short events; promising for non research intensive}

universities

• _{Interoperability test-beds}

_(3%)

–

_{Grids and Clouds;}

_Standards

_{; Open Grid Forum OGF really needs}

• _{Domain Science applications}

_(34%)

–

_{Life sciences highlighted}

_(17%)

• _{Computer science}

_(41%)

–

_{Largest current category}

• _{Computer Systems Evaluation}

_(29%)

–

_{TeraGrid (TIS, TAS, XSEDE), OSG, EGI, Campuses}

• _{Clouds are meant to need less support than other models;}

FutureGrid needs more

user support

…….

(50)

Software Components

• _Portals

_{including “Support” “use FutureGrid” “Outreach”}

• _Monitoring

_{– INCA, Power (GreenIT)}

• _Experiment

_Manager

_{: specify/workflow}

• _Image

_{Generation and Repository}

• _Intercloud

_{Networking ViNE}

• _{Virtual Clusters}

_{built with virtual networks}

• _Performance

_library

• _Rain

_or

_R

_untime

_A

_daptable

_I

_nsertio

_N

_{Service for images}

• _Security

_{Authentication, Authorization,}

“Research”

Above and below

(51)

RAIN related Terminology

• Image Management provides the low level software (create, customize, store, share and deploy images) needed to achieve Dynamic Provisioning and Rain • Abstract Image Management stores templates to create images suitable for

different environments

• Dynamic Provisioning is in charge of providing machines with the requested OS. The requested OS must have been previously deployed in the

infrastructure

(52)

Architecture

API

Image Management

FG Resources

VM Bare Metal_Image

Image Repository Image Generator

External Services: Bcfg2, Security tools Image Deploy

FG Shell Portal

Cloud Framework

(53)

Image Generation

• Creates and customizes

images according to user

:

o

_{OS type}

o

_{OS version}

o

_Architecture

o

_{Software Packages}

• Image is stored in the Image

Repository or returned to the

users

• Images are not aimed to any

specific infrastructure

Command Line Tools

Generate Image Fi x Ba se Im ag e Update Image Verify Image Deployable Base Image Check for Updates Security Checks Store in Image

(54)

Image Deployment

• Customizes images for specific

infrastructures and deploys

them

• Decides if an image is secure

enough to be deployed or if it

needs additional security tests

• Two main infrastructures types

– HPC deployment: Create

network bootable images that

can run in bare metal

machines

– Cloud deployment: Convert

the images in VMs

Customize Image for:

OpenStack Eucalyptus Nimbus OpenNebula Amazon ...

Command Line Tools

Retrieve from Image Repository Deployable

Base Image

Deploy Image in the Infrastructure HPC

Image Customized for the selected Infrastructure

Image is Ready for Instantiation in

(55)

(56)

Image Generation

• Creates and customizes images according to

user requirements

• Images are not aimed to any specific

infrastructure

0 50 100 150 200 250 300 350 400 450 500 CentOS5 Ubuntu10.10 Ti m e (s ) GenerateanImage

Upload image to the repo

Compress image

Install user packages

Install u l packages

Create Base OS

(57)

Image Deployment

• Customizes and deploys images for

specific infrastructures

• Two main infrastructures types:

http://futuregrid.orghttps://portal.futuregrid.org

HPC Deployment Cloud Deployment

0 20 40 60 80 100 120 140 BareMetal Ti m e (s ) Deploy/StageImageonxCAT/Moab xCAT packimage

Retrieve kernels and update xcat tables Untar image and copy to the right place Retrieve image from repo 0 50 100 150 200 250 300 OpenStack Eucalyptus Ti m e (s ) Deploy/StageImageonCloudFrameworks

Wait un l image is in available status (aprox.) Uploading image to cloud framework from client side Retrieve image from server side to client side

Umount image (varies in different execu ons)

Customize image for specific IaaS framework

(58)

Dynamic Provisioning Scalability Test

(59)

1 2 4 8

0 100 200 300 400 500 600 700 800

900 Eucalyptus Deployment

Upload Image to Cloud Framework Retrieve Image from Server Side Customize Image

Number of Concurrent Requests

T im e ( s )

1 2 4 8

0 100 200 300 400 500 600 700 800

900 OpenStack Deployment

Upload Image to Cloud Framework Retrieve Image from Server Side Customize Image

T

im

e

(s

(60)

FutureGrid in a nutshell

• The FutureGrid project

mission

is to

enable experimental work

that advances:

a) Innovation and scientific understanding of

distributed computing and

parallel computing paradigms,

b) The

engineering science of middleware

that enables these paradigms,

c) The use and drivers of these paradigms by

important applications

, and,

d) The

education

of a new generation of students and workforce on the

use of these paradigms and their applications.

• The

implementation

of mission includes

• Distributed flexible hardware

with supported use

• _{Identified IaaS and PaaS “core” software}

_{with supported use}

• Growing list of software from FG partners and users

(61)

EXTRAS

(62)

Genomics in Personal Health

Suppose you measured everybody’s

genome every 2 years

30 petabits of new gene data per day

factor of 100 more for raw reads with coverage

Data surely distributed

1.5*10

8 to 1.5*10

10 continuously running

present day cores to perform a simple Blast

analysis on this data

Amount depends on clever hashing and maybe

Blast not good enough as field gets more

sophisticated

(63)

Clouds and Jobs

• _Clouds

_{are a major industry thrust with a growing fraction of IT expenditure}

that IDC estimates will grow to

$44.2 billion direct investment in 2013

while

15% of IT investment in 2011

will be related to cloud systems with a 30%

growth in public sector.

• _{Gartner also rates cloud computing high on list of critical emerging}

technologies with for example “

Cloud Computing

” and “

Cloud Web

Platforms

” rated as

transformational

(their highest rating for impact) in the

next 2-5 years.

• _{Correspondingly there is and will continue to be major opportunities for}

new jobs in cloud computing

with a recent European study estimating there

will be

2.4 million new cloud computing jobs in Europe alone by 2015

.

• _{Cloud computing spans research and economy and so attractive component}

of

curriculum

for students that mix “going on to PhD” or “graduating and

(64)

https://portal.futuregrid.org ₆₄

Transformational

High

Moderate

Low

Cloud Computing Cloud Web Platforms Media Tablet

“Big Data” and Extreme Information Processing and Management

Cloud Computing

In-memory Database Management Systems

Media Tablet

Cloud/Web Platforms

Private Cloud Computing

QR/Color Bar Code

Social Analytics

(65)

https://portal.futuregrid.org ₆₅

Laptop for PowerPoint

Lego Robot GPS Nokia N800

RFID Tag Cellphones

RFID Reader Surveillance Camera

(66)

Real-Time GPS Sensor Data-Mining

Services process real time data from ~70 GPS

Sensors in Southern California

Brokers and Services on Clouds

– no major

performance issues

66

Streaming Data Support

Transformations Data Checking

Hidden Markov Datamining (JPL)

Display (GIS)

CRTN GPS

Earthquake

Real Time

Archival

(67)

MapReduce and Twister on

Azure

(68)

MapReduceRoles4Azure Architecture

(69)

MapReduceRoles4Azure

• _{Use distributed, highly scalable and highly available cloud services}

as the building blocks.

–

_{Azure Queues}

_{for task scheduling.}

–

_{Azure Blob storage}

_{for input, output and intermediate data storage.}

–

_{Azure Tables}

_{for meta-data storage and monitoring}

• _{Utilize eventually-consistent , high-latency cloud services effectively}

to deliver performance comparable to traditional MapReduce

runtimes.

• _{Minimal management and maintenance overhead}

• _{Supports dynamically scaling up and down of the compute}

resources.

• _{MapReduce fault tolerance}

(70)

Cache aware scheduling

• _{New Job (1}

st

_iteration)

–

_{Through queues}

• _{New iteration}

–

_{Publish entry to Job Bulletin Board}

–

_{Workers pick tasks based on}

in-memory data cache and execution

history (MapTask Meta-Data

cache)

–

_{Any tasks that do not get}

(71)

Performance – Kmeans

Clustering

Number of Executing Map Task Histogram

Strong Scaling with 128M Data Points

(72)

Kmeans Speedup from 32 cores

32 64 96 128 160 192 224 256

0 50 100 150 200 250

Twister4Azure Twister

Hadoop

Number of Cores

R

el

a

ti

ve

S

p

e

ed

u

(73)

AllGather

• _{Broadcasts the Map outputs to all other nodes}

• Assembles them together in the recipient nodes

• _{Schedules the next iteration of the application}

• Eliminates the shuffle, reduce, merge overhead

• Currently implemented using Azure inter-role TCP based all-to-all broadcast

• We have noticed up to 8% speedup

(74)

Twister4Azure Conclusions

• _{Twister4Azure enables users to easily and efficiently}

perform large scale iterative data analysis and scientific

computations on Azure cloud.

–

_{Supports classic and iterative MapReduce}

–

_{Non pleasingly parallel use of Azure}

• _{Utilizes a hybrid scheduling mechanism to provide the}

caching of static data across iterations.

• _{Should integrate with workflow systems}

• _{Plenty of testing and improvements needed!}

• _{Open source: Please use}

(75)

Summary of

Applications Suitable for Clouds

(76)

76

(a) Map Only

(a) Map Only (c) Iterative _Synchronous_Synchronous(d) Loosely (d) Loosely MapReduce (c) Iterative MapReduce (b) Classic MapReduce (b) Classic MapReduce Input Input map map reducereduce Input Input map map reduce reduce Iterations Iterations Input Input Output Output map map P_ij P_ij BLAST Analysis Smith-Waterman Distances Parametric sweeps PolarGrid Matlab data analysis

BLAST Analysis Smith-Waterman Distances

Parametric sweeps PolarGrid Matlab data analysis

High Energy Physics (HEP) Histograms Distributed search Distributed sorting Information retrieval High Energy Physics (HEP) Histograms Distributed search Distributed sorting Information retrieval

Many MPI scientific applications such as solving differential equations and particle dynamics Many MPI scientific applications such as solving differential equations and particle dynamics

Domain of MapReduce and Iterative Extensions

Domain of MapReduce and Iterative Extensions MPIMPI

(77)

Expectation Maximization and

Iterative MapReduce

• _Clustering

_and

_{Multidimensional Scaling}

_{are both EM}

(

expectation maximization

) using deterministic

annealing for improved performance

• _EM

_{tends to be}

_good

_for

_clouds

_and

_Iterative

MapReduce

–

_Quite

_{complicated computations}

_{(so compute largish}

compared to communicate)

–

_{Communication is}

_Reduction

_{operations (global sums in our}

case)

–

_{See also}

_{Latent Dirichlet Allocation}

_{and related Information}

Retrieval algorithms similar structure

(78)

1) A(k) = - 0.5



_i₌₁N



j=1N



(

i

,

j

) <M

i

(

k

)> <M

j

(

k

)> / <C(k)>

2

2) B

_

(k) =



_i₌₁N



₍

_i

_,



_{) <M}

i

(

k

)> / <C(k)>

3) 

_

(

k

) = (B

_

(k) + A(k))

4) <M

_i

(

k

)> = p(k) exp( -



_i

(

k

)/T )/



_k’=1K

_{p(k’) exp(-}



i

(

k’

)/T)

5) C(k) =



_i₌₁N

_<M

i

(

k

)>

6) p(k) = C(k) / N

• _{Loop to converge variables; decrease T from}

_

_;

split centers by halving p(k)

DA-PWC EM Steps (

E is red

, M Black)

k runs over clusters; i,j,



points

78

Steps 1 global sum

(reduction)

(79)

What can we learn?

• _{There are many}

_{pleasingly parallel data analysis}

algorithms which are super for clouds

–

_{Remember SWG computation longer than other parts}

of analysis

• _{There are interesting data mining algorithms}

needing

iterative parallel run times

• _{There are}

_{linear algebra}

_{algorithms with flaky}

compute/communication ratios

• _{Expectation Maximization}

_{good for Iterative}

MapReduce

(80)

Research Issues for (Iterative) MapReduce

• _{Quantify and Extend that Data analysis for Science seems to work well on}

Iterative MapReduce and clouds so far.

– _{Iterative MapReduce (Map Collective)}_{spans all architectures as unifying idea}

• _{Performance and Fault Tolerance Trade-offs;}

– _{Writing to disk each iteration (as in Hadoop) naturally lowers performance but increases}

fault-tolerance

– _{Integration of}_GPU_’s

• _{Security and Privacy technology and policy essential for use in many biomedical}

applications

• _{Storage: multi-user data parallel file systems have scheduling and management}

– _{NOSQL and SciDB on virtualized and HPC systems}

• _{Data parallel Data analysis languages: Sawzall and Pig Latin more successful than}

HPF?

• _{Scheduling: How does research here fit into scheduling built into clouds and}

Iterative MapReduce (Hadoop)

(81)

Architecture of Data-Intensive

Clouds

(82)

Components of a Scientific Computing Platform

Authentication and Authorization: Provide single sign in to All system architectures

Workflow: Support workflows that link job components between Grids and Clouds.

Provenance: Continues to be critical to record all processing and data sources

Data Transport: Transport data between job components on Grids and Commercial Clouds respecting custom storage patterns like Lustre v HDFS

Program Library: Store Images and other Program material

Blob: Basic storage concept similar to Azure Blob or Amazon S3

DPFS Data Parallel File System: Support of file systems like Google (MapReduce), HDFS (Hadoop) or Cosmos (dryad) with compute-data affinity optimized for data processing

Table: Support of Table Data structures modeled on Apache Hbase/CouchDB or Amazon SimpleDB/Azure Table. There is “Big” and “Little” tables – generally NOSQL

SQL: Relational Database

Queues: Publish Subscribe based queuing system

Worker Role: This concept is implicitly used in both Amazon and TeraGrid but was (first) introduced as a high level construct by Azure. Naturally support Elastic Utility Computing MapReduce: Support MapReduce Programming model including Hadoop on Linux, Dryad on Windows HPCS and Twister on Windows and Linux. Need Iteration for Datamining

Software as a Service: This concept is shared between Clouds and Grids

(83)

Architecture of Data Repositories?

• _{Traditionally governments set up repositories for}

data associated with particular missions

–

_{For example EOSDIS, GenBank, NSIDC, IPAC for Earth}

Observation , Gene, Polar Science and Infrared

astronomy

–

_{LHC/OSG computing grids for particle physics}

• _{This is complicated by volume of data deluge,}

distributed instruments as in gene sequencers

(maybe centralize?) and need for complicated

intense computing

(84)

Clouds as Support for Data Repositories?

• _{The data deluge needs cost effective computing}

–

_{Clouds are by definition cheapest}

• _{Shared resources essential (to be cost effective}

and large)

–

_{Can’t have every scientists downloading petabytes to}

personal cluster

• _{Need to reconcile distributed (initial source of )}

data with shared computing

–

_{Can move data to (disciple specific) clouds}

–

_{How do you deal with multi-disciplinary studies}

(85)

Traditional File System?

• _{Typically a shared file system (Lustre, NFS …) used to support high}

performance computing

• _{Big advantages in flexible computing on shared data but doesn’t “}

_bring

computing to data

”

• _{Object stores similar to this?}

(86)

Data Parallel File System?

• _{No archival storage and computing brought to data}

C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data C Data File1 Block1 Block2 BlockN

……

Breakup Replicate each block

File1 Block1 Block2 BlockN

……

Breakup

(87)

Summary of

Data-Intensive Applications on

Clouds

(88)

Summarizing Guiding Principles

• _{Clouds may not be suitable for everything but they are suitable for}

majority of data intensive applications

–

Solving partial differential equations on 100,000 cores probably needs

classic MPI engines

• _{Cost effectiveness, elasticity and quality programming model will}

drive use of clouds in many areas such as genomics

• _{Need to solve issues of}

–

Security-privacy-trust for sensitive data

–

How to store data – “data parallel file systems” (HDFS), Object Stores, or

classic HPC approach with shared file systems with Lustre etc.

• _{Programming model which is likely to be}

_MapReduce

_based

–

_{Look at high level languages}

–

_{Compare with databases (SciDB?)}

–

_{Must support iteration to do “real parallel computing”}

–

_{Need Cloud-HPC Cluster Interoperability}

(89)

FutureGrid in a Nutshell

(90)

What is FutureGrid?

• The FutureGrid project mission is to

enable experimental work

that advances:

a) Innovation and scientific understanding of

distributed computing and

parallel computing paradigms

,

b) The

engineering science of middleware

that enables these paradigms,

c) The use and drivers of these paradigms by

important applications

, and,

d) The

education

of a new generation of students and workforce on the

use of these paradigms and their applications.

• _{The implementation of mission includes}

• _{Distributed flexible hardware}

_{with supported use}

• _Identified

_{IaaS and PaaS}

_{“core” software with supported use}

• Expect growing list of software from FG partners and users

(91)

Motivation for RAIN

• Provide users with the ability to easily

create their own computational

environments (OS, packages, software)

• Users can deploy and run their

environments in both bare-metal and

virtualized infrastructures like Amazon,

OpenStack, Eucalyptus, Nimbus or

(92)

Scalability of Image Generation

• _{Concurrent CentOS image creation requests}

• _{Increasing number of OpenNebula compute nodes to scale}

http://futuregrid.org

1 2 4 8

0 200 400 600 800 1000 1200

1 Compute Node 2 Compute Nodes 4 Compute Nodes

T

im

e

(

s

(93)

HPC Deployment

1 0

20 40 60 80 100 120 140

Packimage (xCAT) Retrieve Kernels and Update xCAT Tables Uncompress Image

Retrieve Image from Reposi-tory

T

im

e

(

s