Google云計(jì)算解決方案5-Hadoop技術(shù)回顧

GoogleClusterComputingFacultyTrainingWorkshopModuleV:HadoopTechnicalReview?SpinnakerLabs,Inc.OverviewHadoopTechnicalWalkthroughHDFSDatabasesUsingHadoopinanAcademicEnvironmentPerformancetipsandothertools?SpinnakerLabs,Inc.YouSay,“tomato…”Googlecallsit:Hadoopequivalent:MapReduceHadoopGFSHDFSBigtableHBaseChubby(nothingyet…butplanned)SomeMapReduceTerminologyJob–A“fullprogram”-anexecutionofaMapperandReduceracrossadatasetTask–AnexecutionofaMapperoraReduceronasliceofdataa.k.a.Task-In-Progress(TIP)TaskAttempt–Aparticularinstanceofanattempttoexecuteataskonamachine?SpinnakerLabs,Inc.TerminologyExampleRunning“WordCount”across20filesisonejob20filestobemappedimply20maptasks+somenumberofreducetasksAtleast20maptaskattemptswillbeperformed…moreifamachinecrashes,etc.?SpinnakerLabs,Inc.TaskAttemptsAparticulartaskwillbeattemptedatleastonce,possiblymoretimesifitcrashesIfthesameinputcausescrashesoverandover,thatinputwilleventuallybeabandonedMultipleattemptsatonetaskmayoccurinparallelwithspeculativeexecutionturnedonTaskIDfromTaskInProgressisnotauniqueidentifier;don’tuseitthatway?SpinnakerLabs,Inc.MapReduce:HighLevel?SpinnakerLabs,Inc.Node-to-NodeCommunicationHadoopusesitsownRPCprotocolAllcommunicationbeginsinslavenodesPreventscircular-waitdeadlockSlavesperiodicallypollfor“status”messageClassesmustprovideexplicitserialization?SpinnakerLabs,Inc.Nodes,Trackers,TasksMasternoderunsJobTrackerinstance,whichacceptsJobrequestsfromclientsTaskTrackerinstancesrunonslavenodesTaskTrackerforksseparateJavaprocessfortaskinstances?SpinnakerLabs,Inc.JobDistributionMapReduceprogramsarecontainedinaJava“jar”file+anXMLfilecontainingserializedprogramconfigurationoptionsRunningaMapReducejobplacesthesefilesintotheHDFSandnotifiesTaskTrackerswheretoretrievetherelevantprogramcode…Where’sthedatadistribution??SpinnakerLabs,Inc.DataDistributionImplicitindesignofMapReduce!Allmappersareequivalent;somapwhateverdataislocaltoaparticularnodeinHDFSIflotsofdatadoeshappentopileuponthesamenode,nearbynodeswillmapinsteadDatatransferishandledimplicitlybyHDFS?SpinnakerLabs,Inc.ConfiguringWithJobConfMRProgramshavemanyconfigurableoptionsJobConfobjectshold(key,value)componentsmappingString’ae.g.,“mapred.map.tasks”20JobConfisserializedanddistributedbeforerunningthejobObjectsimplementingJobConfigurablecanretrieveelementsfromaJobConf?SpinnakerLabs,Inc.WhatHappensInMapReduce?

DepthFirst?SpinnakerLabs,Inc.JobLaunchProcess:ClientClientprogramcreatesaJobConfIdentifyclassesimplementingMapperandReducerinterfacesJobConf.setMapperClass(),setReducerClass()Specifyinputs,outputsJobConf.setInputPath(),setOutputPath()Optionally,otheroptionstoo:JobConf.setNumReduceTasks(),JobConf.setOutputFormat()…?SpinnakerLabs,Inc.JobLaunchProcess:JobClientPassJobConftoJobClient.runJob()orsubmitJob()runJob()blocks,submitJob()doesnotJobClient:DeterminesproperdivisionofinputintoInputSplitsSendsjobdatatomasterJobTrackerserver?SpinnakerLabs,Inc.JobLaunchProcess:JobTrackerJobTracker:InsertsjarandJobConf(serializedtoXML)insharedlocationPostsaJobInProgresstoitsrunqueue?SpinnakerLabs,Inc.JobLaunchProcess:TaskTrackerTaskTrackersrunningonslavenodesperiodicallyqueryJobTrackerforworkRetrievejob-specificjarandconfigLaunchtaskinseparateinstanceofJavamain()isprovidedbyHadoop?SpinnakerLabs,Inc.JobLaunchProcess:TaskTaskTracker.Child.main():SetsupthechildTaskInProgressattemptReadsXMLconfigurationConnectsbacktonecessaryMapReducecomponentsviaRPCUsesTaskRunnertolaunchuserprocess?SpinnakerLabs,Inc.JobLaunchProcess:TaskRunnerTaskRunner,MapTaskRunner,MapRunnerworkinadaisy-chaintolaunchyourMapperTaskknowsaheadoftimewhichInputSplitsitshouldbemappingCallsMapperonceforeachrecordretrievedfromtheInputSplitRunningtheReducerismuchthesame?SpinnakerLabs,Inc.CreatingtheMapperYouprovidetheinstanceofMapperShouldextendMapReduceBaseOneinstanceofyourMapperisinitializedbytheMapTaskRunnerforaTaskInProgressExistsinseparateprocessfromallotherinstancesofMapper–nodatasharing!?SpinnakerLabs,Inc.Mappervoidmap(WritableComparablekey, Writablevalue, OutputCollectoroutput, Reporterreporter)?SpinnakerLabs,Inc.WhatisWritable?Hadoopdefinesitsown“box”classesforstrings(Text),integers(IntWritable),etc.AllvaluesareinstancesofWritableAllkeysareinstancesofWritableComparable?SpinnakerLabs,Inc.WritingForCacheCoherencywhile(moreinputexists){ myIntermediate=newintermediate(input); myIcess(); exportoutputs;}?SpinnakerLabs,Inc.WritingForCacheCoherencymyIntermediate=newintermediate(junk);while(moreinputexists){ myIntermediate.setupState(input); myIcess(); exportoutputs;}?SpinnakerLabs,Inc.WritingForCacheCoherencyRunningtheGCtakestimeReusinglocationsallowsbettercacheusageSpeedupcanbeasmuchastwo-foldAllserializabletypesmustbeWritableanyway,somakeuseoftheinterface?SpinnakerLabs,Inc.GettingDataToTheMapperReadingDataDatasetsarespecifiedbyInputFormatsDefinesinputdata(e.g.,adirectory)IdentifiespartitionsofthedatathatformanInputSplitFactoryforRecordReaderobjectstoextract(k,v)recordsfromtheinputsource?SpinnakerLabs,Inc.FileInputFormatandFriendsTextInputFormat–Treatseach‘\n’-terminatedlineofafileasavalueKeyValueTextInputFormat–Maps‘\n’-terminatedtextlinesof“kSEPv”SequenceFileInputFormat–Binaryfileof(k,v)pairswithsomeadd’lmetadataSequenceFileAsTextInputFormat–Same,butmaps(k.toString(),v.toString())?SpinnakerLabs,Inc.FilteringFileInputsFileInputFormatwillreadallfilesoutofaspecifieddirectoryandsendthemtothemapperDelegatesfilteringthisfilelisttoamethodsubclassesmayoverridee.g.,Createyourown“xyzFileInputFormat”toread*.xyzfromdirectorylist?SpinnakerLabs,Inc.RecordReadersEachInputFormatprovidesitsownRecordReaderimplementationProvides(unused?)capabilitymultiplexingLineRecordReader–ReadsalinefromatextfileKeyValueRecordReader–UsedbyKeyValueTextInputFormat?SpinnakerLabs,Inc.InputSplitSizeFileInputFormatwilldividelargefilesintochunksExactsizecontrolledbymapred.min.split.sizeRecordReadersreceivefile,offset,andlengthofchunkCustomInputFormatimplementationsmayoverridesplitsize–e.g.,“NeverChunkFile”?SpinnakerLabs,Inc.SendingDataToReducersMapfunctionreceivesOutputCollectorobjectOutputCollector.collect()takes(k,v)elementsAny(WritableComparable,Writable)canbeused?SpinnakerLabs,Inc.WritableComparatorComparesWritableComparabledataWillcallWritableCpare()CanprovidefastpathforserializeddataJobConf.setOutputValueGroupingComparator()?SpinnakerLabs,Inc.SendingDataToTheClientReporterobjectsenttoMapperallowssimpleasynchronousfeedbackincrCounter(Enumkey,longamount)setStatus(Stringmsg)Allowsself-identificationofinputInputSplitgetInputSplit()?SpinnakerLabs,Inc.PartitionAndShufflePartitionerintgetPartition(key,val,numPartitions)OutputsthepartitionnumberforagivenkeyOnepartition==valuessenttooneReducetaskHashPartitionerusedbydefaultUseskey.hashCode()toreturnpartitionnumJobConfsetsPartitionerimplementation?SpinnakerLabs,Inc.Reductionreduce( WritableComparablekey, Iteratorvalues, OutputCollectoroutput, Reporterreporter)Keys&valuessenttoonepartitionallgotothesamereducetaskCallsaresortedbykey–“earlier”keysarereducedandoutputbefore“l(fā)ater”keys?SpinnakerLabs,Inc.Finally:WritingTheOutput?SpinnakerLabs,Inc.OutputFormatAnalogoustoInputFormatTextOutputFormat–Writes“keyval\n”stringstooutputfileSequenceFileOutputFormat–Usesabinaryformattopack(k,v)pairsNullOutputFormat–Discardsoutput?SpinnakerLabs,Inc.HDFS?SpinnakerLabs,Inc.HDFSLimitations“Almost”GFSNofileupdateoptions(recordappend,etc);allfilesarewrite-onceDoesnotimplementdemandreplicationDesignedforstreamingRandomseeksdevastateperformance?SpinnakerLabs,Inc.NameNode“Head”interfacetoHDFSclusterRecordsallglobalmetadata?SpinnakerLabs,Inc.SecondaryNameNodeNotafailoverNameNode!Recordsmetadatasnapshotsfrom“real”NameNodeCanmergeupdatelogsinflightCanuploadsnapshotbacktoprimary?SpinnakerLabs,Inc.NameNodeDeathNonewrequestscanbeservedwhileNameNodeisdownSecondarywillnotfailoverasnewprimarySowhyhaveasecondaryatall??SpinnakerLabs,Inc.NameNodeDeath,cont’dIfNameNodediesfromsoftwareglitch,justrebootButifmachineishosed,metadataforclusterisirretrievable!?SpinnakerLabs,Inc.BringingtheClusterBackIforiginalNameNodecanberestored,secondarycanre-establishthemostcurrentmetadatasnapshotIfnot,createanewNameNode,usesecondarytocopymetadatatonewprimary,restartwholecluster()Isthereanotherway…??SpinnakerLabs,Inc.KeepingtheClusterUpProblem:DataNodes“fix”theaddressoftheNameNodeinmemory,can’tswitchinflightSolution:BringnewNameNodeup,butuseDNStomakeclusterbelieveit’stheoriginaloneSecondarycanbethe“new”one?SpinnakerLabs,Inc.FurtherReliabilityMeasuresNamenodecanoutputmultiplecopiesofmetadatafilestodifferentdirectoriesIncludinganNFSmountedoneMaydegradeperformance;watchforNFSlocks?SpinnakerLabs,Inc.Databases?SpinnakerLabs,Inc.LifeAfterGFSStraightGFSfilesarenottheonlystorageoptionHBase(ontopofGFS)providescolumn-orientedstoragemySQLandotherdbenginesstillrelevant?SpinnakerLabs,Inc.HBaseCaninterfacedirectlywithHadoopProvidesitsownInput-andOutputFormatclasses;sendsrowsdirectlytomapper,receivesnewrowsfromreducer…Butmightnotbereadyforclassroomuse(leaststablecomponent)?SpinnakerLabs,Inc.MySQLClusteringMySQLdatabasecanbeshardedonmultipleserversForfastIO,usesamemachinesasHadoopTablescanbesplitacrossmachinesbyrowkeyrangeMultiplereplicascanservesametable?SpinnakerLabs,Inc.Sharding&HadoopPartitionersForbestperformance,ReducershouldgostraighttolocalmysqlinstanceGetalldataintherightmachineinonecopyImplementcustomPartitionertoensureparticularkeyrangegoestomysql-awareReducer?SpinnakerLabs,Inc.AcademicHadoopRequirements?SpinnakerLabs,Inc.ServerProfileUWcluster:40nodes,80processorstotal2GBram/processor24TBrawstoragespace(8TBreplicated)OnenodereservedforJobTracker/NameNodeTwomorewouldn’tcooperate…Butstillvastlyoverpowered?SpinnakerLabs,Inc.Setup&MaintenanceTookabouttwodaystosetupandconfigureMostlyhardware-relatedissuesHadoopsetupwasonlyacouplehoursMaintenance:onlyafewhours/weekMostlyrebootingtheclusterwhenjobsgotstuck?SpinnakerLabs,Inc.TotalUsageAbout15,000CPU-hoursconsumedby20students…Outof130,000availableoverquarterAverageloadisabout12%?SpinnakerLabs,Inc.Analyzingstudentusagepatterns?SpinnakerLabs,Inc.NotQuitetheWholeStoryRealistically,studentsdidmostworkveryclosetodeadlineClustersatunusedforafewdays,followedbyoverloadingfortwodaysstraight?SpinnakerLabs,Inc.AnalyzingstudentusagepatternsLesson:ResourcedemandsareNOTconstant!?SpinnakerLabs,Inc.HadoopJobSchedulingFIFOqueuematchesincomingjobstoavailablenodesNonotionoffairnessNeverswitchesoutrunningjobRun-awaytaskscouldstarveotherstudentjobs?SpinnakerLabs,Inc.HadoopSecurityButonthebright(?)side:NosecuritysystemforjobsAnyonecanstartajob;buttheycanalsocancelotherjobsRealistically,studentsdidnotcancelotherstudentjobs,evenwhentheyshould?SpinnakerLabs,Inc.HadoopSecurity:TheDarkSideNopermissionsinHDFSeitherJustnowaddedin0.16OnestudentdeletedthecommondatasetforaprojectEmailsubject:“Oops…”Nostudentscouldtesttheircodeuntildatasetrestoredfrombackup?SpinnakerLabs,Inc.JobSchedulingLessonsGettingstudentsto“playnice”ishardNoincentiveJustplainbad/buggycodeClustercontentioncausedproblemsatdeadlinesWorkingroupsStaggerdeadlines?SpinnakerLabs,Inc.AnotherPossibilityAmazonEC2provideson-demandserversMaybeabletohavestudentsusetheseforjobs“Labfee”wouldbe~$150/studentSimpleweb-basedinterfacesexistRHadoopOnDemand(HOD)comingsoonInjectsnewnodesintoliveclusters?SpinnakerLabs,Inc.MorePerformance&Scalability?SpinnakerLabs,Inc.NumberofTasksMappers=10*nodes(or3/2*cores)Reducers=2*nodes(or1.05*cores)Twodegreesoffreedominmapperruntime:Numberoftasks/node,andsizeofInputSplitsSee/lucene-hadoop/HowManyMapsAndReduces?SpinnakerLabs,Inc.MorePerformanceTweaksHadoopdefaultstoheapcapof200MBSet:mapred.child.java.opts=-Xmx512m1024MB/processmayalsobeappropriateDFSblocksizeis64MBForhugefiles,setdfs.block.size=134217728mapred.reduce.parallel.copiesSetto15—50;moredata=>morecopies?SpinnakerLabs,Inc.DeadTasksStudentjobswould“runaway”,adminrestartneededVeryoftenstuckinhugeshuffleprocessStudentsdidnotknowaboutPartitionerclass,mayhavehadnon-uniformdistribu