WIXLIB

versityE- ‐mail:panda@cse.ohio- ‐state.eduh p://www.cse.ohio- ‐state.edu/ panda

Introduc'ontoBigDataApplica'onsandAnaly'cs businessanalyFcs ormaFonmanagementanddecisionmaking ddigiFzingmoreinformaFonthanever ’sLaw Commonlyaccepted3V’sofBigData v/itl/ssd/is/upload/NIST- ‐stonebraker.pdf 5V’sofBigData–3V Value,VeracityHPCACSwitzerlandConference(Mar'15)2

DataManagementandProcessingonModernClusters nagementandprocessingsystemsinmulFpleFers– Front- ‐enddataaccessingandserving(Online) Memcached DB(e.g.MySQL),HBase– Back- ‐enddataanalyFcs(Offline) HDFS,MapReduce,SparkFront-end TierInternetWebWebServerWebServerServerData Accessingand Memcached DB(MySQL)Memcached DB(MySQL) DB (MySQL)NoSQL DBNoSQL DB(HBase)NoSQL DB(HBase)(HBase)Back-end TierData Analytics Apps/JobsMapReduceSparkHDFS3

OverviewofApacheHadoopArchitecture Open- TableforBigDataAnalyFcs HadoopCommonUFliFes(RPC,etc.),HDFS,MapReduce,YARN h ystem(HDFS)HadoopCommon/Core(RPC,.)4

SparkArchitectureOverview Anin- ‐memorydata- ntaFonStandalone,YARN,Mesos ScalableandcommunicaFonintensive– s(RDDs)– MapReduce- ‐likeshuffleoperaFonstoreparFFonRDDs– ContextWorkerMasterWorkerhSp://spark.apache.org5

MemcachedArchitectureHighPerformanceNetworks!"# %" memoryCPUsSSDHDDCPUsHDD.(Database Servers)Web Frontend Servers(Memcached Clients).High PUsSSDHDDMainmemoryCPUsSSDHDD.(Memcached Servers) Three- ‐layerarchitectureofWeb2.0– WebServers,MemcachedServers,DatabaseServers DistributedCachingLayer– AllowstoaggregatesparememoryfrommulFplenodes– Generalpurpose lls PCACSwitzerlandConference(Mar'15)6

Presenta'onOutline ChallengesforAcceleraFngBigDataProcessing TheHigh- ‐PerformanceBigData(HiBD)Project RDMA- ‐baseddesignsforApacheHadoopandSpark– CasestudieswithHDFS,MapReduce,andSpark– SamplePerformanceNumbersforRDMA- ‐Hadoop2.x0.9.6Release RDMA- ‐baseddesignsforMemcachedandHBase– RDMA- ‐basedMemcachedwithSSD- ‐assistedHybridMemory– RDMA- ‐basedHBase g– OSUHiBDBenchmarks 7

DriversforModernHPCClusters HighEndCompuFng(HEC)isgrowingdramaFcally– HighPerformanceCompuFng– BigDataCompuFng TechnologyAdvancement– MulF- ‐core/many- ‐coretechnologiesandaccelerators– RemoteDirectMemoryAccess(RDMA)- ‐enablednetworking(InfiniBandandRoCE)– SolidStateDrives(SSDs)andNon- ‐VolaFleRandom- ‐AccessMemory(NVRAM)– anhe–1A8

tch1/10/40GigEIPoIB10/40GigE- ference(Mar'15)9

WideAdop'onofRDMATechnology MessagePassingInterface(MPI)forHPC ParallelFileSystems– Lustre– GPFS Deliveringexcellentperformance:– 1.0microseclatency– 100Gbpsbandwidth– 5- ‐10%CPUuFlizaFon erence(Mar'15)10

ocols?ProtocolCommunica'onandI/OLibraryPoint- ‐to- nVirtualiza'onI/OandFileSystemsQoSFault- ul'- ‐andMany- ference(Mar'15)11

CanBigDataProcessingSystemsbeDesignedwithHigh- Interface1/10GigENetwork10GigEorInfiniBand Socketsnotdesignedforhigh- ‐performance– StreamsemanFcsonenmismatchforupperlayers– Zero- ‐copynotavailablefornon- 5)12

Presenta'onOutline ChallengesforAcceleraFngBigDataProcessing TheHigh- ‐PerformanceBigData(HiBD)Project RDMA- ‐baseddesignsforApacheHadoopandSpark– CasestudieswithHDFS,MapReduce,andSpark– SamplePerformanceNumbersforRDMA- ‐Hadoop2.x0.9.6Release RDMA- ‐baseddesignsforMemcachedandHBase– RDMA- ‐basedMemcachedwithSSD- ‐assistedHybridMemory– RDMA- ‐basedHBase g– OSUHiBDBenchmarks 13

OverviewoftheHiBDProjectandReleases RDMAforApacheHadoop2.x(RDMA- ‐Hadoop- ‐2.x) RDMAforApacheHadoop1.x(RDMA- ‐Hadoop) RDMAforMemcached(RDMA- ‐Memcached) OSUHiBD- ‐Benchmarks(OHB) hSp://hibd.cse.ohio- ‐state.edu UsersBase:95organizaFonsfrom18countries Morethan2,900downloads ureHPCACSwitzerlandConference(Mar'15)14

RDMAforApacheHadoop2.xDistribu'on High- ‐PerformanceDesignofHadoopoverRDMA- ‐enabledInterconnects– pportattheverbs- ‐levelforHDFS,MapReduce,andRPCcomponents– EnhancedHDFSwithin- ‐memoryandheterogeneousstorage– HighperformancedesignofMapReduceoverLustre– HH- ‐M,HHH- (naFveInfiniBand,RoCE,andIPoIB) Currentrelease:0.9.6– BasedonApacheHadoop2.6.0– � Testedwith MellanoxInfiniBandadapters(DDR,QDRandFDR) RoCEsupportwithMellanoxadapters VariousmulF- ‐coreplarorms DifferentfilesystemswithdisksandSSDsandLustre– hSp://hibd.cse.ohio- ‐state.eduHPCACSwitzerlandConference(Mar'15)15

RDMAforMemcachedDistribu'on High- ‐PerformanceDesignofMemcachedoverRDMA- ‐enabledInterconnects– pportattheverbs- ‐levelforMemcachedandlibMemcachedcomponents– radiFonalsockets- � HighperformancedesignofSSD- ‐AssistedHybridMemory Currentrelease:0.9.3– BasedonMemcached1.4.22andlibMemcached1.0.18– CompliantwithlibMemcachedAPIsandapplicaFons– Testedwith ortwithMellanoxadaptersVariousmulF- ‐coreplarormsSSD– hSp://hibd.cse.ohio- ‐state.eduHPCACSwitzerlandConference(Mar'15)16

OSUHiBDMicro- ‐Benchmark(OHB)Suite- ‐Memcached ReleasedinOHB0.7.1(ohb memlat) Evaluatestheperformanceofstand- ‐aloneMemcached Threedifferentmicro- ‐benchmarks– SETMicro- ‐benchmark:Micro- ‐benchmarkformemcachedsetoperaFons– GETMicro- ‐benchmark:Micro- ‐benchmarkformemcachedgetoperaFons– MIXMicro- ‐benchmark:Micro- d:WriteraFois90:10) CalculatesaveragelatencyofMemcachedoperaFons tzerlandConference(Mar'15)17

Presenta'onOutline ChallengesforAcceleraFngBigDataProcessing TheHigh- ‐PerformanceBigData(HiBD)Project RDMA- ‐baseddesignsforApacheHadoopandSpark– CasestudieswithHDFS,MapReduce,andSpark– SamplePerformanceNumbersforHadoop2.x0.9.6Release RDMA- ‐baseddesignsforMemcachedandHBase– RDMA- ‐basedMemcachedwithSSD- ‐assistedHybridMemory– RDMA- ‐basedHBase g– OSUHiBDBenchmarks 18

Accelera'onCaseStudiesandIn- ‐DepthPerformanceEvalua'on RDMA- ‐basedDesignsandPerformanceEvaluaFon– HDFS– MapReduce– SparkHPCACSwitzerlandConference(Mar'15)19

DesignOverviewofHDFSwithRDMA ,RoCE.)– RDMA- ‐basedHDFSwrite– RDMA- ‐basedHDFSreplicaFon– ParallelreplicaFonsupport– On- ‐demandconnecFonsetup– InfiniBand/RoCEsupport FngtradiFonalsocketinterface ywri eninnaFvecodeHPCACSwitzerlandConference(Mar'15)20

IPoIB(QDR)OSU- ‐IB(QDR)2015Reducedby30%10502GB4GB ClusterwithHDDDataNodes6GBFileSize(GB)8GB10GB– 30%improvementincommunicaFonFmeoverIPoIB(QDR)– 56%improvementincommunicaFonFmeover10GigE A- SC),Nov2012Islam,X.Lu,W.Rahman,andD.K.Panda,SOR- ‐HDFS:ASEDA- ‐basedApproachtoMaximizeOverlappinginN.RDMA- nference(Mar'15)21

EnhancedHDFSwithIn- ‐memoryandHeterogeneousStorage DesignFeaturesApplica'ons– ThreemodesTriple- romo'onHeterogeneousStorageRAMDiskSSDHDD Default(HHH) In- ‐Memory(HHH- ‐M) Lustre- ‐Integrated(HHH- ‐L)– devices RAM,SSD,HDD,Lustre– EvicFon/PromoFonbasedondatausagepa ern– HybridReplicaFonLustre– Lustre- ‐Integratedmode: Lustre- ‐basedfault- .K.Panda,Triple- y2015HPCACSwitzerlandConference(Mar'15)22

EnhancedHDFSwithIn- ‐memoryandHeterogeneousStorage–ThreeModes replicaFonschemes– I/OoperaFonsoverRAMdisk,SSD,andHDD– HybridreplicaFon(in- ‐memoryandpersistentstorage)– Be erfault- ‐toleranceaswellasperformance HHH- ‐M:High- ‐performancein- ‐memoryI/OoperaFons– MemoryreplicaFon(in- ‐memoryonlywithlazypersistence)– Asmuchperformancebenefitaspossible HHH- ‐L:Lustreintegrated– TakeadvantageoftheLustreavailableinHPCclusters– Lustre- ‐basedfault- ‐tolerance(NoHDFSreplicaFon)– rence(Mar'15)23

ughput(MBps)5000IPoIB(FDR)OSU- e(s)60004000300020002501501005000readTestDFSIOOSU- 6032:120ClusterSize:DataSizeRandomWriter For160GBTestDFSIOin32nodes For120GBRandomWriterin32– � 3ximprovementoverIPoIB(QDR)– tzerlandConference(Mar'15)24

HHH- ‐L)500450400350300250200150100500HDFS- ‐IPoIB(QDR)Lustre- ‐IPoIB(QDR)OSU- 54%HDFS- ‐IPoIB(QDR)360Lustre- ‐IPoIB(QDR)120OSU- ‐IB(QDR)24060SortStoragespacefor60GBSort For60GBSortin8nodes– 24%improvementoverdefaultHDFS– 54%improvementoverLustre– tzerlandConference(Mar'15)25

Evalua'onwithPUMAandCloudBurst(HHH- ‐L/HHH)Execu'onTime(s)2500Reducedby17%2000HDFS- ‐IPoIB(QDR)Lustre- ‐IPoIB(QDR)OSU- ‐IB(QDR)1500HDFS- ‐IPoIB(FDR)OSU- eCountPUMA PUMAonOSURIGrepCloudBurst CloudBurstonTACCStampede– SequenceCountwithHHH- ‐L:17%benefitoverLustre,8%overHDFS– WithHHH:19%improvementoverHDFS– PCACSwitzerlandConference(Mar'15)26

Accelera'onCaseStudiesandIn- ‐DepthPerformanceEvalua'on RDMA- ‐basedDesignsandPerformanceEvaluaFon– HDFS– MapReduce– SparkHPCACSwitzerlandConference(Mar'15)27

DesignOverviewofMapReducewithRDMA leNetworks– RDMA- ‐basedshuffle– Prefetchingandcachingmapoutput– EfficientShuffleAlgorithms– In- ‐memorymerge– On- ‐demandShuffleAdjustment– Advancedoverlapping map,shuffle,andmerge shuffle,merge,andreduce– On- ‐demandconnecFonsetup– InfiniBand/RoCEsupport(IB,10GE/iWARP,RoCE.) FngtradiFonalsocketinterface ibrarywri eninnaFvecodeHPCACSwitzerlandConference(Mar'15)28

hitecture s– EfficientShuffleAlgorithms– DynamicandEfficientSwitching– On- ighPerformanceInterconnects,ICS,June2014.29

- ‐IB(QDR)1000OSU- 'onTime(sec)12008006004002000IPoIB(FDR)UDA- ‐IB(FDR)OSU- terSize:64SortinOSUCluster For240GBSortin64nodes(512cores)– de For320GBTeraSortin64nodes(1Kcores)– 38%improvementoverIPoIB(FDR)withHDDusedforHDFS30

Evalua'onsusingPUMAWorkload10GigEIPoIB(QDR)OSU- ex(30GB)Benchmarks ize datasizeHPCACSwitzerlandConference(Mar'15)31

ompute NodesApp MasterMapReduceLustre Client HPCClusterDeployment MapReduceoverLustreMetaData ServersObject Storage ServersLustre SetupHPCACSwitzerlandConference(Mar'15)– eparateI/Onodes– ocalstorage– Sub- ems,suchasLustre– Localdiskisusedastheintermediatedatadirectory– Lustreisusedastheintermediatedatadirectory32

ustre DesignFeaturesMap1Map2Map3– Twoshuffleapproaches Lustrereadbasedshuffle treRead/RDMAReduce1In- ‐memorymerge/sortreduceReduce2In- ‐memorymerge/sortreduce– eapproaches– Dynamicallyadaptstothebe �lingvaluesforeachLustrereadoperaFon– In- eptsimilartoRDMA- r'15)33

PerformanceImprovementofMapReduceoverLustreonTACC- ‐Stampede PoIB(FDR)OSU- c)JobExecu'onTime(sec)1000500Reducedby44%OSU- 004005000DataSize(GB) ter:128 For640GBSortin128nodes– 44%improvementoverIPoIB(FDR)– ustre:CanRDMA- ‐basedApproachBenefit?,Euro- 5)34

CaseStudy- SC- ‐Gordon 'onTime(sec)800700600IPoIB(QDR)OSU- ‐Lustre- ‐Read(QDR)OSU- ‐RDMA- ‐IB(QDR)OSU- ‐Hybrid- 0500400300200700600IPoIB(QDR)OSU- ‐Lustre- ‐Read(QDR)OSU- ‐RDMA- ‐IB(QDR)OSU- ‐Hybrid- ‐IB(QDR)Reducedby25%5004003002001001000040604080 For120GBTeraSortin16nodes– ence(Mar'15)120DataSize(GB)DataSize(GB) For80GBSortin8nodes80– 25%improvementoverIPoIB(QDR)35

Accelera'onCaseStudiesandIn- ‐DepthPerformanceEvalua'on RDMA- ‐basedDesignsandPerformanceEvaluaFon– HDFS– MapReduce– SparkHPCACSwitzerlandConference(Mar'15)36

DesignOverviewofSparkwithRDMA DesignFeaturesSpark rk(Scala/Java)BlockManagerJava ava onal)RDMAShuffleFetcher(plug-in)Java SocketRDMA-based Shuffle Engine(Java/JNI)1/10 Gig Ethernet/IPoIB (QDR/FDR)NetworkNative InfiniBand(QDR/FDR)– RDMAbasedshuffle– SEDA- ‐basedplugins– DynamicconnecFonmanagementandsharing– Non- ‐blockingandout- ‐of- ‐orderdatatransfer– Off- ‐JVM- ‐heapbuffermanagement– InfiniBand/RoCEsupport FngtradiFonalsocketinterface arywri ndConference(Mar'15)37

PreliminaryResultsofSpark- ‐RDMADesign- upBywith64cores ttasks– 18%improvementoverIPoIB(QDR)for10GBdatasize ttasks– itzerlandConference(Mar'15)38

Presenta'onOutline ChallengesforAcceleraFngBigDataProcessing TheHigh- ‐PerformanceBigData(HiBD)Project RDMA- ‐baseddesignsforApacheHadoopandSpark– CasestudieswithHDFS,MapReduce,andSpark– SamplePerformanceNumbersforHadoop2.x0.9.6Release RDMA- ‐baseddesignsforMemcachedandHBase– RDMA- ‐basedMemcachedwithSSD- ‐assistedHybridMemory– RDMA- ‐basedHBase g– OSUHiBDBenchmarks 39

)7000Increasedby7.8x6000300IPoIB(FDR)OSU- 50Execu'onTime(s)TotalThroughput(MBps)8000OSU- IOwithatotal128maps ForThroughput,– 6- ‐7.8ximprovementoverIPoIBfor80- ) ForLatency,– 2.5- ‐3ximprovementoverIPoIBfor80- ‐120GBfilesize40