Challenges in Big Data, Big Simulations, Clouds and HPC

(1)

1

Challenges in Big Data,

Big Simulations, Clouds and HPC

Geoffrey Fox

July 19, 2016

[email protected]

http://www.dsc.soic.indiana.edu/, http://spidal.org/ http://hpc-abds.org/kaleidoscope/

Department of Intelligent Systems Engineering

School of Informatics and Computing, Digital Science Center Indiana University Bloomington

IEEE Cloud and Big Data Computing

Toulouse France July 18-21

(2)

Abstract

• _{We review several questions at the intersection of}

_Big

Data, Big Simulations, Clouds and HPC

. We

consider broad topics:

–

_{Application Requirements}

–

_{Software (including language)}

–

_{System (hardware)}

–

_{Focus on Parallel processing in applications}

• _{We consider these in context of a}

_{Big Data - Big}

Simulation convergence

and propose a simple

approach which we are pursuing although it is

certainly not agreed

2

(3)

What are the application and user

requirements?

e.g. is the data streaming, how similar are

commercial and scientific requirements?

NIST Big Data Initiative

Use Cases and Properties

(4)

51 Detailed Use Cases:

Contributed July-September 2013

Covers goals, data features such as 3 V’s, software, hardware

• _{Government Operation(4):}_{National Archives and Records Administration, Census Bureau} • _{Commercial(8):}_{Finance in Cloud, Cloud Backup, Mendeley (Citations), Netflix, Web Search,}

Digital Materials, Cargo shipping (as in UPS)

• _Defense(3):_{Sensors, Image surveillance, Situation Assessment}

• _{Healthcare and Life Sciences(10):}_{Medical records, Graph and Probabilistic analysis,}

Pathology, Bioimaging, Genomics, Epidemiology, People Activity models, Biodiversity

• _{Deep Learning and Social Media(6):}_{Driving Car, Geolocate images/cameras, Twitter, Crowd}

Sourcing, Network Science, NIST benchmark datasets

• _{The Ecosystem for Research(4):}_{Metadata, Collaboration, Language Translation, Light source}

experiments

• _{Astronomy and Physics(5):}_{Sky Surveys including comparison to simulation, Large Hadron}

Collider at CERN, Belle Accelerator II in Japan

• _{Earth, Environmental and Polar Science(10):}_{Radar Scattering in Atmosphere, Earthquake,}

Ocean, Earth Observation, Ice sheet Radar scattering, Earth radar mapping, Climate simulation datasets, Atmospheric turbulence identification, Subsurface Biogeochemistry (microbes to

watersheds), AmeriFlux and FLUXNET gas sensors

• _Energy(1):_{Smart grid}

• _{Published by NIST as}_{http://nvlpubs.nist.gov/nistpubs/SpecialPublications/NIST.SP.1500-3.pdf} • _{“Version 2” being prepared}

4

02/16/2016

(5)

5

02/16/2016 http://hpc-abds.org/kaleidoscope/survey/

(6)

Sample Features of 51 Use Cases I

• _{PP (26)}

_“All”

_{Pleasingly Parallel or Map Only}

• _{MR (18)}

_{Classic MapReduce MR (add MRStat below for full count)}

• _{MRStat (7}

_{) Simple version of MR where key computations are simple}

reduction as found in statistical averages such as histograms and

averages

• _{MRIter (23}

₎

_{Iterative MapReduce or MPI (Flink, Spark, Twister)}

• _{Graph (9)}

_{Complex graph data structure needed in analysis}

• _{Fusion (11)}

_{Integrate diverse data to aid discovery/decision making;}

could involve sophisticated algorithms or could just be a portal

• _{Streaming (41)}

_{Some data comes in incrementally and is processed}

this way

• _{Classify (30)}

_{Classification: divide data into categories}

• _{S/Q (12)}

_{Index, Search and Query}

(7)

Sample Features of 51 Use Cases II

• _{CF (4)}_{Collaborative Filtering for recommender engines}

• _{LML (36)}_{Local Machine Learning (Independent for each parallel entity) –}

application could have GML as well

• _{GML (23)}_{Global Machine Learning: Deep Learning, Clustering, LDA, PLSI,}

MDS,

– _{Large Scale Optimizations as in Variational Bayes, MCMC, Lifted Belief}

Propagation, Stochastic Gradient Descent, L-BFGS, Levenberg-Marquardt . Can call EGO or Exascale Global Optimization with scalable parallel algorithm

• _{Workflow (51)}_Universal

• _{GIS (16)}_{Geotagged data and often displayed in ESRI, Microsoft Virtual}

Earth, Google Earth, GeoServer etc.

• _HPC ₍₅₎_{Classic large-scale simulation of cosmos, materials, etc.}

generating (visualization) data

• _{Agent (2)}_{Simulations of models of data-defined macroscopic entities}

represented as agents

(8)

Data and Model in Big Data and Simulations

• _{Need to discuss}

_Data

_and

_Model

_{as problems combine them,}

but we can get insight by separating which allows better

understanding of

Big Data - Big Simulation “convergence”

(or differences!)

• _{Big Data}

_{implies Data is large but Model varies}

– _e.g._LDA_{with many topics or}_{deep learning}_{has large model}

– _{Clustering or Dimension reduction can be quite small in model size}

• _Simulations

_{can also be considered as}

_Data

_and

_Model

– _Model_{is solving particle dynamics or partial differential equations} – _Data_{could be small when just boundary conditions}

– _Data_{large with data assimilation (weather forecasting) or when data}

visualizations are produced by simulation

• _Data

_{often static between iterations (unless streaming);}

_Model

varies between iterations

8

(9)

Classifying Use Cases

• _{Take 51 NIST and other use cases}

_

_{derive multiple specific}

features

• _{Generalize and systematize with features termed “facets”}

• _{50 Facets (Big Data) termed Ogres}

_{divided into 4 sets or}

views where each view has “similar” facets

• _{Add simulations and look separately at}

_{Data and Model}

_gives

64 Facets

describing

Big Simulation and Data

termed

Convergence Diamonds

looking at either

data or model

or

their combination

• _{Allows one to study coverage of benchmark sets and}

architectures

9

(10)

64 Features in 4 views for Unified Classification of Big Data

and Simulation Applications

10 Lo ca l ( A na ly ti cs /I nf or m ati cs /S im ul ati o ns ) 2 M

Data Source and Style View

Pleasingly Parallel Classic MapReduce Map-Collective Map Point-to-Point Shared Memory Single Program Multiple Data Bulk Synchronous Parallel Fusion Dataflow Agents Workflow

Geospatial Information System HPC Simulations

Internet of Things Metadata/Provenance

Shared / Dedicated / Transient / Permanent Archived/Batched/Streaming – S1, S2, S3, S4, S5 HDFS/Lustre/GPFS

Files/Objects

Enterprise Data Model SQL/NoSQL/NewSQL 1 M M ic ro -b e nc h m ar ks Execution View Processing View 1 2 3 4 6 7 8 9 10 11M 12 10D 9 8D 7D 6D 5D 4D 3D 2D 1D

Map Streaming 5

Convergence Diamonds Views and

Facets

Problem Architecture View

15 M C or e Li br ar ie s V is ua liz a ti on 14 M G ra p h A lg or it h m s 13 M Li ne ar A lg e br a Ke rn e ls /M a ny s u bc la ss e s 12 M G lo ba l ( A na ly ti cs /I nf or m ati cs /S im ul ati o ns ) 3 M R ec om m e nd e r E ng in e 5 M 4 M B as e D at a St a ti sti cs 10 M St re am in g D a ta A lg or it hm s O pti m iz a ti o n M e th od o lo gy 9 M Le ar n in g 8 M D a ta C la ss ifi ca ti o n 7 M D at a Se ar ch /Q ue ry /I nd ex 6 M 11 M D a ta A li gn m en t

Big Data Processing Diamonds M u lti sc al e M et ho d 17 M 16 M It er ati ve P D E S ol ve rs 22 M N at ur e of m es h if u se d Ev ol uti on o f D is cr e te S ys te m s 21 M Pa rti cl e s an d Fi el ds 20 M N -b od y M e th od s 19 M Sp e ct ra l M et ho d s 18 M Simulation (Exascale) Processing Diamonds D a ta A b st ra cti o n D 12 M o de l A bs tra cti on M 12 D at a M e tri c = M / N on -M e tri c = N D 13 D at a M et ric = M / N on -M et ric = N M 13 ܱ ܰ ଶ = N N / ܱ ሺ ܰ ሻ = N M 14 R eg u la r= R / Irr e gu la r = I M o de l M 10 Ve ra cit y 7 Ite ra ti ve / Sim p le M 11 C om m un ic ati o n St ru ct u re M 8 D yn am ic = D / St ati c = S D 9 D yn a m ic = D / St ati c = S M 9 R e gu la r = R / Irr eg ul ar = I D a ta D 10 M o de l V ar ie ty M 6 D at a Ve lo cit y D 5 Pe rfo rm an ce M et ric s 1 D a ta V ar ie ty D 6 Flo ps p er B yt e /M e m or y IO /F lo ps p er w att 2 Ex e cu ti on En vir on m en t; C or e lib ra rie s 3 D a ta Vo lu m e D 4 M od el Siz e M 4 Simulations Analytics (Model for Big Data)

Both

(All Model)

(Nearly all Data+Model)

(Nearly all Data)

(Mix of Data and Model)

(11)

Convergence Diamonds and their 4 Views I

• _One

_view

_{is the overall}

_{problem architecture or}

macropatterns

which is naturally related to the machine

architecture needed to support application.

–

_Unchanged

_{from Ogres and describes properties of problem}

such as “Pleasing Parallel” or “Uses Collective

Communication”

• _The

_{execution (computational) features or micropatterns}

view

, describes issues such as I/O versus compute rates,

iterative nature and regularity of computation and the classic

V’s of Big Data: defining problem size, rate of change, etc.

–

_{Significant changes}

_{from ogres to separate}

_Data

_and

(12)

Convergence Diamonds and their 4 Views II

• _The_{data source & style} _view_{includes facets specifying how the data is}

collected, stored and accessed. Has classic database characteristics

– _{Simulations can have facets here to describe input or output data}

– _Examples:_Streaming_{, files versus objects, HDFS v. Lustre}

• _Processing _view_{has model (not data) facets which describe types of}

processing steps including nature of algorithms and kernels by model e.g. Linear Programming, Learning, Maximum Likelihood, Spectral methods, Mesh type,

– _{mix of Big Data Processing View and Big Simulation Processing View}

and includes some facets like “uses linear algebra” needed in both: has specifics of key simulation kernels and in particular includes facets seen in NAS Parallel Benchmarks and Berkeley Dwarfs

• _{Instances of Diamonds}_{are particular problems and a set of Diamond}

instances that cover enough of the facets could form a comprehensive benchmark/mini-app set

(13)

DISCUSSION

What are the application and user requirements?

e.g. is the data streaming, how similar are

commercial and scientific requirements?

NIST Big Data Initiative

Use Cases and Properties

(14)

Structure of Applications

• _Real-time₍_streaming_{) data is increasingly common in scientific and}

engineering research, and it is ubiquitous in commercial Big Data (e.g., social network analysis, recommender systems and consumer behavior classification)

– _{So far little use of commercial and Apache technology in analysis of}

scientific streaming data

• _{Pleasingly parallel}_{applications important in science (long tail) and data}

communities

• _{Commercial-Science application differences:}_{Search and recommender}

engines have different structure to deep learning, clustering, topic models, graph analyses such as subgraph mining

– _{Latter very sensitive to communication and can be hard to parallelize} – _{Search typically not as important in Science as in commercial use as}

search volume scales by number of users

• _{Should discuss}_data_and_model_separately

– _{Term data often used rather sloppily and often refers to model}

14

(15)

What is structure of problems and what does

this imply?

e.g. do big data requirements imply clouds or HPC machines or both e.g. difference between big data and big simulation problem structure

The Big Data Ogres and Convergence Diamonds

6 Forms of MapReduce

2 Forms of Communication/Flow

(16)

Problem Architecture

View (Meta or MacroPatterns)

i. Pleasingly Parallel – as in BLAST, Protein docking, some (bio-)imagery including Local Analytics or Machine Learning – ML or filtering pleasingly parallel, as in bio-imagery, radar images (pleasingly parallel but sophisticated local analytics)

ii. Classic MapReduce: Search, Index and Query and Classification algorithms like collaborative filtering (G1 for MRStat in Features, G7)

iii. Map-Collective: Iterative maps + communication dominated by “collective” operations as in reduction, broadcast, gather, scatter. Common datamining pattern

iv. Map-Point to Point: Iterative maps + communication dominated by many small point to point messages as in graph algorithms

v. Map-Streaming: Describes streaming, steering and assimilation problems

vi. Shared Memory: Some problems are asynchronous and are easier to parallelize on shared rather than distributed memory – see some graph algorithms

vii. SPMD: Single Program Multiple Data, common parallel programming feature

viii. BSP or Bulk Synchronous Processing: well-defined compute-communication phases ix. Fusion: Knowledge discovery often involves fusion of multiple methods.

x. Dataflow: Important application features often occurring in composite Ogres xi. Use Agents: as in epidemiology (swarm approaches) This is Model only xii. Workflow: All applications often involve orchestration (workflow) of multiple

components

16

02/16/2016

(17)

Local and Global Machine Learning

• _{Many applications}

_use

_{LML or Local machine Learning}

_where

machine learning (often from R or Python or Matlab) is run

separately on every data item such as on every image

• _{But others}

_are

_GML

_{Global Machine Learning where machine}

learning is a basic algorithm run over all data items (over all nodes

in computer)

–

_{maximum likelihood or}



2

with a sum over the N data items –

documents, sequences, items to be sold, images etc. and often

links (point-pairs).

–

_{GML includes Graph analytics, clustering}

_/community

detection, mixture models, topic determination, Multidimensional

scaling, (

Deep

)

Learning Networks

• _{Note Facebook may need lots of small graphs (one per person and}

~LML) rather than one giant graph of connected people (GML)

(18)

6 Forms of

MapReduce

Describes

Architecture of

- Problem (Model reflecting data) - Machine - Software 2 important variants (software) of Iterative MapReduce and Map-Streaming a) “In-place” HPC

b) Flow for model and data

18 5/17/2016

1) Map-Only

Pleasingly Parallel

4) Map- Point to Point Communication

3) Iterative MapReduce or Map-Collective

-2) Classic MapReduce Input map reduce Input map reduce Iterations Input Output map Local Graph 5) Map-Streaming maps _brokers Events 6) Shared-Memory Map Communication Shared Memory

Map & Communication

1) Map-Only Pleasingly Parallel

4) Map- Point to Point Communication 3) Iterative MapReduce

or Map-Collective

-2) Classic MapReduce Input map reduce Input map reduce Iterations Input Output map Local Graph 5) Map-Streaming maps _brokers Events 6) Shared-Memory Map Communication Shared Memory

(19)

Relation of Problem and Machine Architecture

• _Problem_is_{Model plus Data}

• _{In my old papers (especially book Parallel Computing Works!), I discussed}

computing as multiple complex systems mapped into each other

Problem



Numerical formulation



Software



Hardware

• _{Each of these 4 systems has an architecture that can be described in}

similar language

• _{One gets an easy programming model if architecture of problem matches}

that of Software

• _{One gets good performance if architecture of hardware matches that of}

software and problem

• _{So “MapReduce” can be used as architecture of software (programming}

model) or “Numerical formulation of problem”

19

(20)

Comparison of Data Analytics with Simulation I

• _{Simulations (models)}_produce_{big data}_{as visualization of results – they}

are data source

– _Or_{consume often smallish data to define a simulation problem} – _{HPC simulation}_{in (weather) data assimilation is}_{data + model} • _{Pleasingly parallel}_{often important in both}

• _{Both are often}_SPMD_and_BSP

• Non-iterative MapReduce is major big data paradigm

– not a common simulation paradigm except where “Reduce” summarizes pleasingly parallel execution as in some Monte Carlos

• _{Big Data often has}_{large collective communication}

– Classic simulation has a lot of smallish point-to-point messages – Motivates Map-Collective model

• _{Simulations characterized often by}_difference_{or differential operators}

leading to nearest neighbor sparsity

• _{Some important data analytics can be sparse as in PageRank and “Bag of words”}

algorithms but many involve full matrix algorithm

(21)

“Force Diagrams” for macromolecules and Facebook

(22)

Comparison

of Data Analytics with Simulation II

• _{There are similarities between some}_{graph problems and particle simulations}_with

a strange cutoff force.

– _Both_{Map-Communication}

• _{Note many big data problems are “}_{long range force}_{” (as in gravitational simulations)}

as all points are linked.

– _{Easiest to parallelize. Often full matrix algorithms}

– _{e.g. in DNA sequence studies, distance}₍_i_,_j_{) defined by BLAST,}

Smith-Waterman, etc., between all sequences i, j.

– _{Opportunity for “fast multipole” ideas in big data. See NRC report}

• _{Current Ogres/Diamonds do not have facets to designate}_{underlying hardware}_:

GPU v. Many-core (Xeon Phi) v. Multi-core as these define how maps processed; they keep map-X structure fixed; maybe should change as ability to exploit vector or SIMD parallelism could be a model facet.

• _{In image-based}_{deep learning}_{, neural network weights are block sparse}

(corresponding to links to pixel blocks) but can be formulated as full matrix operations on GPUs and MPI in blocks.

• _In_{HPC benchmarking}_{, Linpack being challenged by a new sparse conjugate}

gradient benchmark HPCG, while I am diligently using non- sparse conjugate gradient solvers in clustering and Multi-dimensional scaling.

(23)

What about the many choices for

infrastructure and middleware?

Should we use classic HPC cluster, Docker or OpenStack (OpenNebula)? Do we need dataflow or more like MPI and SPMD, Bulk Synchronous

Processing?

Should we use threads or processes?

Where are Big Data (Apache) approaches superior/inferior to those familiar from Grid and HPC work?

Software Functionality and Performance

HPC-ABDS

High performance Computing Enhanced Apache Big Data Stack

Clouds v. HPC clusters

Flink, Spark, HPC(MPI) Tradeoffs

(24)

24 5/17/2016

Kaleidoscope of (Apache) Big Data Stack (ABDS) and HPC Technologies Green is HPC work of NSF14-43054

Cross-Cutting Functions

1) Message and Data Protocols: Avro, Thrift, Protobuf

2) Distributed Coordination : Google Chubby, Zookeeper, Giraffe, JGroups

3) Security & Privacy: InCommon, Eduroam OpenStack Keystone, LDAP, Sentry, Sqrrl, OpenID, SAML OAuth 4) Monitoring: Ambari, Ganglia, Nagios, Inca

17) Workflow-Orchestration: ODE, ActiveBPEL, Airavata, Pegasus, Kepler, Swift, Taverna, Triana, Trident, BioKepler, Galaxy, IPython, Dryad, Naiad, Oozie, Tez, Google FlumeJava, Crunch, Cascading, Scalding, e-Science Central, Azure Data Factory, Google Cloud Dataflow, NiFi (NSA), Jitterbit, Talend, Pentaho, Apatar, Docker Compose, KeystoneML

16) Application and Analytics: Mahout , MLlib , MLbase, DataFu, R, pbdR, Bioconductor, ImageJ, OpenCV, Scalapack, PetSc, PLASMA MAGMA, Azure Machine Learning, Google Prediction API & Translation API, mlpy, scikit-learn, PyBrain, CompLearn, DAAL(Intel), Caffe, Torch, Theano, DL4j, H2O, IBM Watson, Oracle PGX, GraphLab, GraphX, IBM System G, GraphBuilder(Intel), TinkerPop, Parasol, Dream:Lab, Google Fusion Tables, CINET, NWB, Elasticsearch, Kibana, Logstash, Graylog, Splunk, Tableau, D3.js, three.js, Potree, DC.js, TensorFlow, CNTK

15B) Application Hosting Frameworks: Google App Engine, AppScale, Red Hat OpenShift, Heroku, Aerobatic, AWS Elastic Beanstalk, Azure, Cloud Foundry, Pivotal, IBM BlueMix, Ninefold, Jelastic, Stackato, appfog, CloudBees, Engine Yard, CloudControl, dotCloud, Dokku, OSGi, HUBzero, OODT, Agave, Atmosphere

15A) High level Programming: Kite, Hive, HCatalog, Tajo, Shark, Phoenix, Impala, MRQL, SAP HANA, HadoopDB, PolyBase, Pivotal HD/Hawq, Presto, Google Dremel, Google BigQuery, Amazon Redshift, Drill, Kyoto Cabinet, Pig, Sawzall, Google Cloud DataFlow, Summingbird, Lumberyard

14B) Streams: Storm, S4, Samza, Granules, Neptune, Google MillWheel, Amazon Kinesis, LinkedIn, Twitter Heron, Databus, Facebook Puma/Ptail/Scribe/ODS, AzureStream Analytics, Floe, Spark Streaming, Flink Streaming, DataTurbine

14A) Basic Programming model and runtime, SPMD, MapReduce: Hadoop, Spark, Twister, MR-MPI, Stratosphere (Apache Flink), Reef, Disco, Hama, Giraph, Pregel, Pegasus, Ligra, GraphChi, Galois, Medusa-GPU, MapGraph, Totem

13) Inter process communication Collectives, point-to-point, publish-subscribe: MPI, HPX-5, Argo BEAST HPX-5 BEAST PULSAR, Harp, Netty, ZeroMQ, ActiveMQ, RabbitMQ, NaradaBrokering, QPid, Kafka, Kestrel, JMS, AMQP, Stomp, MQTT, Marionette Collective, Public Cloud: Amazon SNS, Lambda, Google Pub Sub, Azure Queues, Event Hubs

12) In-memory databases/caches: Gora (general object from NoSQL), Memcached, Redis, LMDB (key value), Hazelcast, Ehcache, Infinispan, VoltDB, H-Store

12) Object-relational mapping: Hibernate, OpenJPA, EclipseLink, DataNucleus, ODBC/JDBC

12) Extraction Tools: UIMA, Tika

11C) SQL(NewSQL): Oracle, DB2, SQL Server, SQLite, MySQL, PostgreSQL, CUBRID, Galera Cluster, SciDB, Rasdaman, Apache Derby, Pivotal Greenplum, Google Cloud SQL, Azure SQL, Amazon RDS, Google F1, IBM dashDB, N1QL, BlinkDB, Spark SQL

11B) NoSQL: Lucene, Solr, Solandra, Voldemort, Riak, ZHT, Berkeley DB, Kyoto/Tokyo Cabinet, Tycoon, Tyrant, MongoDB, Espresso, CouchDB, Couchbase, IBM Cloudant, Pivotal Gemfire, HBase, Google Bigtable, LevelDB, Megastore and Spanner, Accumulo, Cassandra, RYA, Sqrrl, Neo4J, graphdb, Yarcdata, AllegroGraph, Blazegraph, Facebook Tao, Titan:db, Jena, Sesame

Public Cloud: Azure Table, Amazon Dynamo, Google DataStore

11A) File management: iRODS, NetCDF, CDF, HDF, OPeNDAP, FITS, RCFile, ORC, Parquet

10) Data Transport: BitTorrent, HTTP, FTP, SSH, Globus Online (GridFTP), Flume, Sqoop, Pivotal GPLOAD/GPFDIST

9) Cluster Resource Management: Mesos, Yarn, Helix, Llama, Google Omega, Facebook Corona, Celery, HTCondor, SGE, OpenPBS, Moab, Slurm, Torque, Globus Tools, Pilot Jobs

8) File systems: HDFS, Swift, Haystack, f4, Cinder, Ceph, FUSE, Gluster, Lustre, GPFS, GFFS Public Cloud: Amazon S3, Azure Blob, Google Cloud Storage

7) Interoperability: Libvirt, Libcloud, JClouds, TOSCA, OCCI, CDMI, Whirr, Saga, Genesis

6) DevOps: Docker (Machine, Swarm), Puppet, Chef, Ansible, SaltStack, Boto, Cobbler, Xcat, Razor, CloudMesh, Juju, Foreman, OpenStack Heat, Sahara, Rocks, Cisco Intelligent Automation for Cloud, Ubuntu MaaS, Facebook Tupperware, AWS OpsWorks, OpenStack Ironic, Google Kubernetes, Buildstep, Gitreceive, OpenTOSCA, Winery, CloudML, Blueprints, Terraform, DevOpSlang, Any2Api

5) IaaS Management from HPC to hypervisors: Xen, KVM, QEMU, Hyper-V, VirtualBox, OpenVZ, LXC, Linux-Vserver, OpenStack, OpenNebula, Eucalyptus, Nimbus, CloudStack, CoreOS, rkt, VMware ESXi, vSphere and vCloud, Amazon, Azure, Google and other public Clouds

(25)

Functionality of 21 HPC-ABDS Layers

1) Message Protocols:

2) Distributed Coordination:

3) Security & Privacy:

4) Monitoring:

5) IaaS Management from HPC to

hypervisors:

6) DevOps:

7) Interoperability:

8) File systems:

9) Cluster Resource

Management:

10) Data Transport:

11) A) File management

B) NoSQL

C) SQL

25 02/16/2016

12) In-memory databases & caches /

Object-relational mapping / Extraction

Tools

13) Inter process communication

Collectives, point-to-point,

publish-subscribe, MPI:

14) A) Basic Programming model and

runtime, SPMD, MapReduce:

B) Streaming:

15) A) High level Programming:

B) Frameworks

(26)

Parallel Computing Approaches

• _{Both simulations and data analytics use similar}_{parallel computing}_ideas • _{Both do}_{decomposition}_{of both model and data}

• _{Both tend use}_SPMD_{and often use}_BSP_{Bulk Synchronous Processing} • _{One has computing (called}_maps_{in big data terminology) and}

communication/reduction (more generally collective) phases

• _{Big data}_{thinks of problems as}_{multiple linked queries}_{even when queries}

are small and uses dataflow model

• _Simulation_{uses dataflow for multiple linked applications but small steps just}

as iterations are done in place

• _Reduction_{in HPC (}_MPIReduce_{) done as optimized tree or pipelined}

communication between same processes that did computing

• _Reduction_{in Hadoop or Flink done as separate processes using dataflow} – _{This leads to 2 forms (}_In-Place_and_Flow_{) of}_Map-X_{described earlier}

• _Interesting_{Fault Tolerance}_{issues highlighted by Hadoop-MPI comparisons}

– not discussed here!

(27)

Programming Model I

• _{Programs are broken up into parts}

–

_{Functionally (coarse grain)}

–

_{Data/model parameter decomposition (fine grain)}

27 5/17/2016

Possible Iteration

Dataflow

MPI

• Fine grain

needs low

latency or

minimal data

copying

• Coarse grain

has lower

(28)

• _MPI

_{designed for fine grain case and typical of parallel computing}

used in large scale simulations

–

_{Only change in model parameters}

_{are transmitted}

• _Dataflow

_{typical of distributed or Grid computing paradigms}

–

_{Data sometimes and model parameters certainly transmitted}

–

_{Caching in iterative MapReduce avoids data communication and}

in fact systems like TensorFlow, Spark or Flink are called dataflow

but usually implement

“model-parameter” flow

• _Different

_{Communication/Compute ratios}

_{seen in different cases}

with ratio (measuring overhead) larger when grain size smaller.

Compare

–

_{Intra-job reduction such as Kmeans}

_{clustering accumulation of}

center changes at end of each iteration and

–

_Inter-Job

_{Reduction as at end of a}

_query

_{or word count operation}

28 5/17/2016

(29)

• _{Need to distinguish}

–

_{Grain size}

_and

_{Communication/Compute ratio}

_{(characteristic of}

problem or component (iteration) of problem)

–

_DataFlow

_versus

_{“Model-parameter” Flow}

_{(characteristic of}

algorithm)

–

_In-Place

_versus

_Flow

_{Software implementations}

• _{Inefficient to use same mechanism independent of characteristics}

• _{Classic Dataflow is approach of Spark and Flink so need to add}

parallel in-place computing as done by

Harp for Hadoop

–

_TensorFlow

_{uses In-Place technology}

• _{Note parallel machine learning (GML not LML) ca}

_{n benefit from}

HPC style interconnects

and

architectures

as seen in GPU-based

deep learning

–

_{So commodity clouds not necessarily best}

29 5/17/2016

(30)

Kmeans Clustering Flink and MPI

one million points fixed; various # centers

24 cores on 16 nodes

30 5/17/2016

Flink

(31)

Harp (Hadoop Plugin) brings HPC to ABDS

• _{Basic Harp: Iterative HPC communication; scientific data abstractions} • _{Careful support of distributed data AND distributed model}

• _{Avoids parameter server approach but distributes model over worker nodes}

and supports collective communication to bring global model to each node

• _{Applied first to Latent Dirichlet Allocation LDA with large model and data}

31 5/17/2016

Shuffle M M M M

Collective Communication

M M M M

R R

MapCollective Model MapReduce Model

YARN MapReduce V2

Harp MapReduce

Applications

(32)

Automatic parallelization

• _{Database community looks at big data job as a dataflow of (SQL)}_queries

and filters

• _{Apache projects like Pig, MRQL and Flink aim at}_{automatic query} optimization by dynamic integration of queries and filters including iteration and different data analytics functions

• _{Going back to ~1993, High Performance Fortran HPF compilers optimized}

set of array and loop operations for large scale parallel execution of optimized vector and matrix operations

• _{HPF worked fine}_{for initial simple regular applications but ran into trouble}

for cases where parallelism hard (irregular, dynamic)

• _{Will same happen in Big Data world?}

• _{Straightforward to}_{parallelize k-means clustering}_{but sophisticated}

algorithms like Elkans method (use triangle inequality) and fuzzy clustering are much harder (but not used much NOW)

• _Will_{Big Data technology run into HPF-style trouble}_{with growing use}

of sophisticated data analytics?

(33)

The choice of language for Big Data and Big

Simulation?

C++, Java, Scala, Python, R highlights performance v. productivity trade-offs.

What is actual performance of Big Data implementations and what are good benchmarks?

Need more Performance evaluation

(34)

Java MPI performs better than FJ Threads I

34 02/16/2016

• 48 24 core Haswell nodes 200K DA-MDS Dataset size • Default MPI much worse than threads

• Optimized MPI using shared memory node-based messaging is much better than threads (default OMPI does not support SM for needed collectives)

All MPI

(35)

Java MPI performs better than FJ Threads II

128 24 core Haswell nodes on SPIDAL DA-MDS Code

35

02/16/2016

Best FJ Threads intra node; MPI inter node

Best MPI; inter and intra node

MPI; inter/intra node; Java not optimized

Speedup compared to 1

(36)

36

1x24x16 2x12x16 3x8x16 4x6x16 6x4x16 8x3x16 12x2x16 24x1x16

0 5000 10000 15000 20000 25000 30000 35000 40000 45000 50000

Case 0: FJ proc-bound thread-bound Case 1: LRT proc-unbound thread-unbound Case 2: LRT proc-bound thread-bound Case 3: LRT proc-unbound thread-bound

Parallel Pattern -- T x P x N where T is threads per process, P is process per node, and N is number of nodes.

Ti

m

e

(m

s)

C0 C1 C2 C3 C4 C5 C6 C7

Socket 0 Socket 1

P0,P1..,P7

MPI, Fork-Join and Long Running

Threads

_{K-Means 1 million 2D points and 1k centers}

• _{Case 0: FJ threads} – _{Proc bound to T}

cores using MPI

– _{Threads inherit the}

same binding

• _{Case 1: LRT threads} – _{Procs and threads}

are both unbound

• _{Case 2: LRT threads} – _{Similar to Case 0}

with LRT threads

• _{Case 3: LRT threads} – _{Proc bound to all}

cores (= non bound)

– _{Each worker thread}

is bound to a core

Fork Join

All MPI

All Threads

Differ in

helper

threads

(37)

DISCUSSION

Is software sustainability important?

Is the Apache model a good approach to this? How useful is DevOps to specify software systems?

Some but not dramatic pressure on HPC community

DevOps promising but not mature

(38)

“Software Engineering”

• _{Fixed Stacks or a “sea of components”?}_{There are ongoing trade-offs}

between “integrated” systems and those involving collections of components. Funding availability important here

– _{Commercial big data tends to use the latter with the component of}

choice changing rapidly (e.g., as seen in Hadoop replaced by Spark and now Spark being challenged by Flink).

– _{Conversely, the HPC community remains committed to much of the}

1990s tool base.

– _{Performance focus suggest fixed stacks to HPC}

• _{Implications of HPC Isolation:}_{The fraction of students learning traditional}

HPC languages and tools is declining rapidly. If the HPC developer base is to be sustained, HPC must embrace more widely used tools

38

(39)

Performance, Productivity,

Sustainability

• _{Performance-Productivity:}

_{The Big Data community}

emphasizes rapid development and human productivity, even

at the expense of execution efficiency. The HPC community

has long discussed productivity but has not yet embraced it as

a metric of success.

• _{Performance Focus of HPC:}

_{The earlier “Grid computing”}

emphasis of HPC computing did not have same performance

emphasis seen in big simulation work today (i.e. it was nearer

Big Data in philosophy).

• _{Sustainability:}

_{The economic sustainability model for Big Data}

software seems more robust than that for big simulation, given

the relative sizes of the communities and the rapid growth of

data analytics.

39

(40)

Big Data - Big Simulation

Convergence?

HPC-Clouds convergence?

Where are differences in approach, opinion?

Convergence Diamonds

HPC-ABDS Software on differently optimized hardware

infrastructure

(41)

• _{Applications, Benchmarks and Libraries}

– _{51 NIST Big Data Use Cases, 7 Computational Giants of the NRC Massive Data Analysis,}

13 Berkeley dwarfs, 7 NAS parallel benchmarks

– _{Unified discussion by separately discussing}_{data & model}_{for each application;} – _{64 facets– Convergence Diamonds -- characterize applications}

– _{Characterization identifies hardware and software features for each application across}

big data, simulation; “complete” set of benchmarks (NIST)

• _{Software Architecture and its implementation}

– _HPC-ABDS:_{Cloud-HPC interoperable software: performance of HPC (High Performance}

Computing) and the rich functionality of the Apache Big Data Stack.

– _{Added HPC}_{to Hadoop, Storm, Heron, Spark; will add to Beam and Flink} – _{Work in Apache model contributing code}

• _{Run same HPC-ABDS across all}_{platforms but “data management” nodes have different}

balance in I/O, Network and Compute from “model” nodes

– _{Optimize to data and model functions as specified by convergence diamonds} – _{Do not optimize for simulation and big data}

• _{Convergence Language:}_{Make C++, Java, Scala, Python … perform well}

Components in Big Data HPC

Convergence

(42)

Typical Convergence Architecture

• _Running_{same HPC-ABDS software}_{across all platforms but data management}

machine has different balance in I/O, Network and Compute from “model” machine

• _{Model has similar issues}_{whether from}_{Big Data or Big Simulation}_.

42 02/16/2016 C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D C D

C

(43)

DISCUSSION

Big Data - Big Simulation

Convergence?

HPC-Clouds convergence?

Where are differences in approach, opinion?

Convergence Diamonds

HPC-ABDS Software on differently optimized hardware

infrastructure

(44)

What do we mean by

convergence?

• _{Commercial v. Science Data:}_{Big Data reasonably clearly defined but covers}

the major commercial activities as well as scientific data analysis; the commercial activity is much the largest and we typically refer to that in discussing convergence. However analyzing scientific big data is very important and must be considered

• _{Super large v Typical HPC Clusters:}_{HPC spans both "leadership class}

supercomputers" and general HPC clusters from departmental size and above. Most deep learning for Big Data operates (today) on medium scale hardware. Hence, the current HPC synergy is mostly on

departmental/university scale machines. Starting with leadership class systems is a much harder problem (culturally and technically), as it is

equivalent to seeking convergence between commercial data centers and leadership class machines.

– _{Note possible relevance of capacity versus capability distinction with}

most observational data analysis on HPC clusters using machine in capacity mode.

44

(45)

Socio-Technical Issues in

Convergence I

• _{Change is Hard I:}_{A lot of scientific data analysis (e.g. astronomy and particle}

physics) started a long time ago and established processes before many recent Big Data developments occurred.

• _{Change is Hard II:}_{SQL Databases have been around a long but many aspects of a}

modern Big Data stack (Hadoop, R, NoSQL, machine learning) are very recent

• _{Interdisciplinary collaboration}_{is critical to make progress; users and}

technologists, industry, academic and government.

– _{In some areas it still needs to be improved – see below!}

• _{HPC & Database Communities:}_{Big Data arose from the database and computer}

science communities; HPC and scientific computing arose from the science and engineering community. They are distinct cultures.

• _Expertise:_{It could be helpful to fold in discussion of data science and}

education/training. Some choices today may be impacted by lack of broad expertise of computing staff in some Big Data issues.