軟件工程基礎(chǔ)Hadoop生態(tài)系統(tǒng)劉馳軟件工程基礎(chǔ)1AnEcosystemforCloudComputing222ProblemBatch(offline)processingofhugedatasetusingcommodityhardwareisnotenoughforreal-timeapplicationsStrongdesireforlinearscalabilityNeedinfrastructuretohandleallmechanicsallowdeveloperstofocusontheprocessinglogic/algorithms3ProblemBatch(offline)process3ExplosiveData!–StorageNewYorkStockExchange:1TBdataperdayFacebook:100billionphotos,1PB(1000TB)InternetArchive:2PBdata,growingby20TBpermonthCan’tputdataonaSINGLEnodeStrongneedsfordistributedfilesystemsExplosiveData!–StorageNewY45Java/Python/Cinterfaces5Java/Python/Cinterfaces5666CommercialHardware典型的2層構(gòu)架–節(jié)點(diǎn)是普通的商業(yè)PC機(jī)–30-40節(jié)點(diǎn)/Rack–頂層到Rack帶寬3-4Gbps–Rack到節(jié)點(diǎn)帶寬1Gbps7CommercialHardware典型的2層構(gòu)架77Whois(was)UsingHadoop?8Whois(was)UsingHadoop?88Example:Facebook的Hadoop集群產(chǎn)品集群4800個內(nèi)核，600個機(jī)器，每個機(jī)器16GB—2009年4月8000個內(nèi)核，1000個機(jī)器，每個機(jī)器32GB—2009年7月每個機(jī)器擁有4個1TB大小的SATA硬盤兩層網(wǎng)絡(luò)結(jié)構(gòu)，每個Rack有40個機(jī)器整個集群大小為2PB，未來還會不斷增加測試集群800個內(nèi)核,每個16GB9Example:Facebook的Hadoop集群產(chǎn)品集群9ADistributedFileSystem軟件工程基礎(chǔ)Hadoop生態(tài)系統(tǒng)劉馳AnEcosystemforCloudComputing課件10Single-NodeArchitectureMemoryDiskCPUMachineLearning,Statistics“Classical”DataMining11Single-NodeArchitectureMemory11CommodityClustersWebdatasetscanbeverylargeTenstohundredsofTBCannotmineonasingleserverStandardarchitectureemerging:ClusterofcommodityLinuxnodesGigabitEthernetinterconnectHowtoorganizecomputationsonthisarchitecture?Maskissuessuchashardwarefailure12CommodityClustersWebdataset12ClusterArchitectureMemDiskCPUMemDiskCPU…SwitchEachrackcontains16-64nodesMemDiskCPUMemDiskCPU…SwitchSwitch1Gbpsbetweenanypairofnodesinarack2-10Gbpsbackbonebetweenracks13ClusterArchitectureMemDiskCPU13StablestorageFirstorderproblem:ifnodescanfail,howcanwestoredatapersistently?Answer:DistributedFileSystemProvidesglobalfilenamespaceGoogleGFS;HadoopHDFS;KosmixKFSTypicalusagepatternHugefiles(100sofGBtoTB)DataisrarelyupdatedinplaceReadsandappendsarecommon14StablestorageFirstorderprob14151515161616NamenodeandDatanodesMaster/slavearchitecture1Namenode,amasterserverthatmanagesthefilesystemnamespaceandregulatesaccesstofilesbyclients.manyDataNodesusuallyonepernodeinacluster.managestorageservesread,writerequests,performsblockcreation,deletion,andreplicationuponinstructionfromNamenode.HDFSexposesafilesystemnamespaceandallowsuserdatatobestoredinfiles.AfileissplitintooneormoreblocksandsetofblocksarestoredinDataNodes.2023/9/2717NamenodeandDatanodesMaster/s17Namespace2023/9/27HierarchicalfilesystemwithdirectoriesandfilesCreate,remove,move,renameetc.NamenodemaintainsthefilesystemAnymetainformationchangestothefilesystemrecordedbytheNamenode.Anapplicationcanspecifythenumberofreplicasofthefileneeded:replicationfactorofthefile.ThisinformationisstoredintheNamenode.18Namespace2023/8/8Hierarchical18DataReplication2023/9/27Storeverylargefilesacrossmachinesinalargecluster.Eachfileisasequenceofblocksofsamesize.Blocksarereplicated2-3times.Blocksizeandreplicasareconfigurableperfile.NamenodereceivesaHeartbeatandaBlockReportfromeachDataNodeinthecluster.BlockReportcontainsalltheblocksonaDatanode.19DataReplication2023/8/8Store19ReplicaPlacement2023/9/27Rack-aware:Goal:improvereliability,availabilityandnetworkbandwidthutilizationResearchtopicNamenodedeterminestherackidforeachDataNode.Replicasareplaced:1inalocalrack,1onadifferentnodeinthelocalrackand1onanodeinadifferentrack.1/3ofthereplicaonanode,2/3onarackand1/3distributedevenlyacrossremainingracks.20ReplicaPlacement2023/8/8Rack-20HDFS:DataNodeDistance21HDFS:DataNodeDistance2121ReplicationPipeliningWhentheclientreceivesresponsefromNamenode,itflushesitsblockinsmallpieces(4K)tothefirstreplica,thatinturncopiesittothenextreplicaandsoon.ThusdataispipelinedfromDatanodetothenext.2023/9/2722ReplicationPipeliningWhenthe22ReplicaSelection2023/9/27ReplicaselectionforREADoperation:HDFStriestominimizethebandwidthconsumptionandlatency.IfthereisareplicaontheReadernodethenthatispreferred.HDFSclustermayspanmultipledatacenters:replicainthelocaldatacenterispreferredovertheremoteone.23ReplicaSelection2023/8/8Repl23Datanode2023/9/27ADatanodestoresdatainfilesinitslocalfilesystem.DatanodehasnoknowledgeaboutHDFSfilesystemItstoreseachblockofHDFSdatainaseparatefile.Datanodedoesnotcreateallfilesinthesamedirectory.Itusesheuristicstodetermineoptimalnumberoffilesperdirectoryandcreatesdirectoriesappropriately:Researchissue?WhenthefilesystemstartsupitgeneratesalistofallHDFSblocksandsendthisreporttoNamenode:Blockreport.24Datanode2023/8/8ADatanodesto24HDFS:FileRead25HDFS:FileRead2525HDFS:FileWrite26HDFS:FileWrite2626CommunicationProtocol2023/9/27AllprotocolsarelayeredontopoftheTCP/IPprotocolAclientestablishesaconnectiontoaconfigurableTCPportontheNamenodemachine.IttalksClientProtocolwiththeNamenode.DatanodestalktotheNamenodeusingDatanodeprotocol.RPCabstractionwrapsbothClientProtocolandDatanodeprotocol.Namenodeissimplyaserverandneverinitiatesarequest;itonlyrespondstoRPCrequestsissuedbyDataNodesorclients.27CommunicationProtocol2023/8/827DataNodeFailureandHeartbeatDatanodesloseconnectivitywithNamenode.NamenodedetectsthisconditionbytheabsenceofaHeartbeatmessage.NamenodemarksDatanodeswithoutHearbeatanddoesnotsendanyIOrequeststothem.AnydataregisteredtothefailedDatanodeisnotavailabletotheHDFS.2023/9/2728DataNodeFailureandHeartbeat28ClusterRebalancingHDFSarchitectureiscompatiblewithdatarebalancingschemes.AschememightmovedatafromoneDatanodetoanotherifthefreespaceonaDatanodefallsbelowacertainthreshold.Intheeventofasuddenhighdemandforaparticularfile,aschememightdynamicallycreateadditionalreplicasandrebalanceotherdatainthecluster.Thesetypesofdatarebalancingarenotyetimplemented:researchissue.2023/9/2729ClusterRebalancingHDFSarchit29APIsHDFSprovidesJavaAPIforapplicationtouse.Pythonaccessisalsousedinmanyapplications.AClanguagewrapperforJavaAPIisalsoavailable.AHTTPbrowsercanbeusedtobrowsethefilesofaHDFSinstance.2023/9/2730APIsHDFSprovidesJavaAPIfor30FSShell,AdminandBrowserInterfaceHDFSorganizesitsdatainfilesanddirectories.ItprovidesacommandlineinterfacecalledtheFSshellthatletstheuserinteractwithdataintheHDFS.Thesyntaxofthecommandsissimilartobashandcsh.Example:tocreateadirectory/foodir/bin/hadoopdfs–mkdir/foodirThereisalsoDFSAdmininterfaceavailableBrowserinterfaceisalsoavailabletoviewthenamespace.2023/9/2731FSShell,AdminandBrowserIn31ADistributedComputationFrameworkforBatchProcessing軟件工程基礎(chǔ)Hadoop生態(tài)系統(tǒng)劉馳AnEcosystemforCloudComputing課件32WhatisMap/Reduce?AProgrammingModelDecomposeaprocessingjobintoMapandReducestagesDeveloperneedtoprovidecodesforMapandReducefunctionsconfigurethejobletHadoophandletherest33WhatisMap/Reduce?AProgrammi33MapReduceModel34MapReduceModel3434DistributedExecutionOverviewUserProgramWorkerWorkerMasterWorkerWorkerWorkerforkforkforkassignmapassignreducereadlocalwriteremoteread,sortOutputFile0OutputFile1writeSplit0Split1Split2InputData35DistributedExecutionOverview35Example:WordCountWehavealargefileofwords,onewordtoalineCountthenumberofappearancesforeachdistinctwordSampleapplication:analyzewebserverlogstofindpopularURLs36Example:WordCountWehaveal36Pseudo-Code:WordCountmap(key,value)://key:documentname;value:textofdocument foreachwordwinvalue: emit(w,1)reduce(key,values)://key:aword;values:aniteratorovercounts result=0 foreachcountvinvalues: result+=v emit(key,result)37Pseudo-Code:WordCountmap(key37map(key=url,val=contents):Foreachwordwincontents,emit(w,“1”)reduce(key=word,values=uniq_counts):Sumall“1”sinvalueslistEmitresult“(word,sum)”seebobrunseespotthrowsee 1bob 1run 1see 1spot 1throw 1bob 1run 1see 2spot 1throw 138WordCountmap(key=url,val=contents):see38MapReduceInput:asetofkey/valuepairsUsersuppliestwofunctions:map(k,v)list(k1,v1)reduce(k1,list(v1))v2(k1,v1)isanintermediatekey/valuepairOutputisthesetof(k1,v2)pairs39MapReduceInput:asetofkey/v39WhatisMAP?

Mapeachdataentryintoapair<key,value>ExamplesMapeachlogfileentryinto<URL,1>Mapdaystocktradingrecordinto<STOCK,Price>40WhatisMAP?Mapeachdataent40WhatisShuffle/Mergephase?

Hadoopmerges(shuffles)outputoftheMAPstageinto:<key,valulue1,value2,value3>Examples<URL,1,1,1,1,11><STOCK,PriceOnday1,PriceOnday2,…...>41WhatisShuffle/Mergephase?H41WhatisReduce?

ReduceentriesproducesbyHadoopmergingprocessinginto<key,value>pairExamplesMap<URL,1,1,1>into<URL,3>Map<Stock,3,2,10>into<Stock,10>42WhatisReduce?Reduceentries42Exampleuses:distributedgrep

distributedsort

weblink-graphreversalterm-vector/hostwebaccesslogstatsinvertedindexconstructiondocumentclusteringmachinelearningstatisticalmachinetranslation.........WidelyApplicable

MapReduceProgramsinGoogleSourceTree43Exampleuses:distributedgrep43100s/1000sof2-CPUx86machines,2-4GBofmemoryLimitedbisectionbandwidthStorageisonlocalIDEdisksGFS:distributedfilesystemmanagesdata(SOSP'03)Jobschedulingsystem:jobsmadeupoftasks, schedulerassignstaskstomachines

ImplementationisaC++librarylinkedintouserprograms

ImplementationOverview44ImplementationOverview4444ImplementationOverviewJobtrackerTasktrackerTasktrackerTasktrackerMasterNodeSlavenode1Slavenode2SlavenodeNWorkersuserWorkersWorkers45ImplementationOverviewJobtra45HowItWorks?46HowItWorks?4646DataFlowInput,finaloutputarestoredonHDFSSchedulertriestoschedulemaptasks“close”tophysicalstoragelocationofinputdataIntermediateresultsarestoredonlocalFSofmapandreduceworkersOutputisofteninputtoanothermapreducetask47DataFlowInput,finaloutputa47CoordinationMasterdatastructuresTaskstatus:(idle,in-progress,completed)IdletasksgetscheduledasworkersbecomeavailableWhenamaptaskcompletes,itsendsthemasterthelocationandsizesofitsRintermediatefiles,oneforeachreducerMasterpushesthisinfotoreducersMasterpingsworkersperiodicallytodetectfailures48CoordinationMasterdatastruct48FailuresMapworkerfailureMaptaskscompletedorin-progressatworkerareresettoidleReduceworkersarenotifiedwhentaskisrescheduledonanotherworkerReduceworkerfailureOnlyin-progresstasksareresettoidleMasterfailureMapReducetaskisabortedandclientisnotified49FailuresMapworkerfailure4949Execution

50Execution5050ParallelExecution51ParallelExecution5151HowManyMapandReduceJobs?Mmaptasks,RreducetasksRuleofthumb:M,R>>(#ofnodes)inclusterOneDFSchunkpermapiscommonImprovesdynamicloadbalancingandspeedsrecoveryfromworkerfailureUsuallyRissmallerthanM,becauseoutputisspreadacrossRfiles52HowManyMapandReduceJobs?M52CombinersOftenamaptaskwillproducemanypairsoftheform(k,v1),(k,v2),…forthesamekeyke.g.,popularwordsinWordCountCansavenetworktimebypre-aggregatingatmappercombine(k1,list(v1))v2sameasreducefunction53CombinersOftenamaptaskwill53PartitionFunctionInputstomaptasksarecreatedbycontiguoussplitsofinputfileForreduce,weneedtoensurethatrecordswiththesameintermediatekeyendupatthesameworkerSystemcanuseadefaultpartitionfunctione.g.,hash(key)modRSometimesusefultooverridee.g.,hash(hostname(URL))modRensuresURLsfromahostendupinthesameoutputfile54PartitionFunctionInputstoma54ExecutionSummaryHowisthisdistributed?Partitioninputkey/valuepairsintochunks,runmap()tasksinparallelAfterallmap()sarecomplete,consolidateallemittedvaluesforeachuniqueemittedkeyNowpartitionspaceofoutputmapkeys,andrunreduce()inparallelIfmap()orreduce()fails,re-execute!55ExecutionSummaryHowisthisd55Example:TradingDataProcessingInput:Historica