大规模数据处理云计算Introduction

大规模数据处理/云计算

Lecture3–MapReduceBasics闫宏飞北京大学信息科学技术学院7/12/2011/~course/cs402/ThisworkislicensedunderaCreativeCommonsAttribution-Noncommercial-ShareAlike3.0UnitedStates

See/licenses/by-nc-sa/3.0/us/fordetailsJimmyLinUniversityofMaryland课程建设SEWMGroupHowdowescaleup?Source:Wikipedia(IBMRoadrunner)DivideandConquer“Work”w1w2w3r1r2r3“Result”“worker”“worker”“worker”PartitionCombineParallelizationChallengesHowdoweassignworkunitstoworkers?Whatifwehavemoreworkunitsthanworkers?Whatifworkersneedtosharepartialresults?Howdoweaggregatepartialresults?Howdoweknowalltheworkershavefinished?Whatifworkersdie?Whatisthecommonthemeofalloftheseproblems?CommonTheme?Parallelizationproblemsarisefrom:Communicationbetweenworkers(e.g.,toexchangestate)Accesstosharedresources(e.g.,data)Thus,weneedasynchronizationmechanismSource:RicardoGuimarãesHerrmannManagingMultipleWorkersDifficultbecauseWedon’tknowtheorderinwhichworkersrunWedon’tknowwhenworkersinterrupteachotherWedon’tknowtheorderinwhichworkersaccessshareddataThus,weneed:Semaphores(lock,unlock)Conditionalvariables(wait,notify,broadcast)BarriersStill,lotsofproblems:Deadlock,livelock,raceconditions...Diningphilosophers,sleepybarbers,cigarettesmokers...Moralofthestory:becareful!CurrentToolsProgrammingmodelsSharedmemory(pthreads)Messagepassing(MPI)DesignPatternsMaster-slavesProducer-consumerflowsSharedworkqueuesMessagePassingP1P2P3P4P5SharedMemoryP1P2P3P4P5MemorymasterslavesproducerconsumerproducerconsumerworkqueueWheretherubbermeetstheroadConcurrencyisdifficulttoreasonaboutConcurrencyisevenmoredifficulttoreasonaboutAtthescaleofdatacenters(evenacrossdatacenters)InthepresenceoffailuresIntermsofmultipleinteractingservicesNottomentiondebugging…Thereality:Lotsofone-offsolutions,customcodeWriteyouowndedicatedlibrary,thenprogramwithitBurdenontheprogrammertoexplicitlymanageeverythingSource:Wikipedia(FlatTire)Source:MITOpenCoursewareSource:MITOpenCoursewareSource:Harper’s(Feb,2008)What’sthepoint?It’sallabouttherightlevelofabstractionThevonNeumannarchitecturehasserveduswell,butisnolongerappropriateforthemulti-core/clusterenvironmentHidesystem-leveldetailsfromthedevelopersNomoreraceconditions,lockcontention,etc.SeparatingthewhatfromhowDeveloperspecifiesthecomputationthatneedstobeperformedExecutionframework(“runtime”)handlesactualexecutionThedatacenteristhecomputer!“BigIdeas”Scale“out”,not“up”LimitsofSMPandlargeshared-memorymachinesMoveprocessingtothedataClusterhavelimitedbandwidthProcessdatasequentially,avoidrandomaccessSeeksareexpensive,diskthroughputisreasonableSeamlessscalabilityFromthemythicalman-monthtothetradablemachine-hourMapReducegggggfffffMapFoldRootsinFunctionalProgrammingTypicalLarge-DataProblemIterateoveralargenumberofrecordsExtractsomethingofinterestfromeachShuffleandsortintermediateresultsAggregateintermediateresultsGeneratefinaloutputKeyidea:provideafunctionalabstractionforthesetwooperationsMapReduce(DeanandGhemawat,OSDI2004)19MapReduceProgrammersspecifytwofunctions:map(k,v)→<k’,v’>*reduce(k’,v’)→<k’’,v’’>*AllvalueswiththesamekeyaresenttothesamereducerTheexecutionframeworkhandleseverythingelse…20mapmapmapmapShuffleandSort:aggregatevaluesbykeysreducereducereducek1k2k3k4k5k6v1v2v3v4v5v6ba12cc36ac52bc78a15b27c2368r1s1r2s2r3s321MapReduceProgrammersspecifytwofunctions:map(k,v)→<k’,v’>*reduce(k’,v’)→<k’,v’>*AllvalueswiththesamekeyaresenttothesamereducerTheexecutionframeworkhandleseverythingelse…What’s“everythingelse”?22MapReduce“Runtime”HandlesschedulingAssignsworkerstomapandreducetasksHandles“datadistribution”MovesprocessestodataHandlessynchronizationGathers,sorts,andshufflesintermediatedataHandleserrorsandfaultsDetectsworkerfailuresandrestartsEverythinghappensontopofadistributedFS(later)23MapReduceProgrammersspecifytwofunctions:map(k,v)→<k’,v’>*reduce(k’,v’)→<k’,v’>*AllvalueswiththesamekeyarereducedtogetherTheexecutionframeworkhandleseverythingelse…Notquite…usually,programmersalsospecify:partition(k’,numberofpartitions)→partitionfork’Oftenasimplehashofthekey,e.g.,hash(k’)modnDividesupkeyspaceforparallelreduceoperationscombine(k’,v’)→<k’,v’>*Mini-reducersthatruninmemoryafterthemapphaseUsedasanoptimizationtoreducenetworktraffic24combinecombinecombinecombineba12c9ac52bc78partitionpartitionpartitionpartitionmapmapmapmapk1k2k3k4k5k6v1v2v3v4v5v6ba12cc36ac52bc78ShuffleandSort:aggregatevaluesbykeysreducereducereducea15b27c298r1s1r2s2r3s3c236825Twomoredetails…BarrierbetweenmapandreducephasesButwecanbegincopyingintermediatedataearlierKeysarriveateachreducerinsortedorderNoenforcedorderingacrossreducers26“HelloWorld”:WordCountMap(Stringdocid,Stringtext):

foreachwordwintext:Emit(w,1);Reduce(Stringterm,Iterator<Int>values):

intsum=0;foreachvinvalues:sum+=v;Emit(term,sum);27MapReducecanreferto…TheprogrammingmodelTheexecutionframework(aka“runtime”)ThespecificimplementationUsageisusuallyclearfromcontext!28MapReduceImplementationsGooglehasaproprietaryimplementationinC++BindingsinJava,PythonHadoopisanopen-sourceimplementationinJavaAnApacheprojectLargecontributionofdevelopmentledbyYahoo,usedinproductionRapidlyexpandingsoftwareecosystemLotsofcustomresearchimplementationsForGPUs,cellprocessors,etc.29split0split1split2split3split4workerworkerworkerworkerworkerMasterUser

Programoutputfile0outputfile1(1)submit(2)schedulemap(2)schedulereduce(3)read(4)localwrite(5)remoteread(6)writeInputfilesMapphaseIntermediatefiles(onlocaldisk)ReducephaseOutputfilesAdaptedfrom(DeanandGhemawat,OSDI2004)30Howdowegetdatatotheworkers?ComputeNodesNASSANWhat’stheproblemhere?31DistributedFileSystemDon’tmovedatatoworkers…moveworkerstothedata!StoredataonthelocaldisksofnodesintheclusterStartuptheworkersonthenodethathasthedatalocalWhy?NotenoughRAMtoholdallthedatainmemoryDiskaccessisslow,butdiskthroughputisreasonableAdistributedfilesystemistheanswerGFS(GoogleFileSystem)forGoogle’sMapReduceHDFS(HadoopDistributedFileSystem)forHadoop32GFS:AssumptionsCommodityhardwareover“exotic”hardwareScale“out”,not“up”HighcomponentfailureratesInexpensivecommoditycomponentsfailallthetime“Modest”numberofhugefilesMulti-gigabytefilesarecommon,ifnotencouragedFilesarewrite-once,mostlyappendedtoPerhapsconcurrentlyLargestreamingreadsoverrandomaccessHighsustainedthroughputoverlowlatencyGFSslidesadaptedfrommaterialby(Ghemawatetal.,SOSP2003)33GFS:DesignDecisionsFilesstoredaschunksFixedsize(64MB)ReliabilitythroughreplicationEachchunkreplicatedacross3+chunkserversSinglemastertocoordinateaccess,keepmetadataSimplecentralizedmanagementNodatacachingLittlebenefitduetolargedatasets,streamingreadsSimplifytheAPIPushsomeoftheissuesontotheclient(e.g.,datalayout)HDFS=GFSclone(samebasicideas)34FromGFStoHDFSTerminologydifferences:GFSmaster=HadoopnamenodeGFSchunkservers=HadoopdatanodesFunctionaldifferences:NofileappendsinHDFS(plannedfeature)HDFSperformanceis(likely)slowerForthemostpart,we’llusetheHadoopterminology…35Adaptedfrom(Ghemawatetal.,SOSP2003)(filename,blockid)(blockid,blocklocati