HIC_2011_Realtime_Analytics_at_Facebook

Real-timeAnalyticsatFacebook:DataFreewayandPumaZhengShao12/2/20111AnalyticsandReal-time2DataFreeway3Puma4FutureWorksAgendaAnalyticsandReal-timewhatandwhyFacebookInsights•Usecases▪Websites/Ads/Apps/Pages▪Timeseries▪Demographicbreak-downs▪Uniquecounts/heavyhitters•Majorchallenges▪Scalability▪LatencyAnalyticsbasedonHadoop/Hive•3000-nodeHadoopcluster•Copier/Loader:Map-Reducehidesmachinefailures•PipelineJobs:HiveallowsSQL-likesyntax•Goodscalability,butpoorlatency!24–48hours.ScribeNFSHTTPHiveHadoopMySQLsecondssecondsHourlyCopier/LoaderDailyPipelineJobsHowtoGetLowerLatency?•Small-batchProcessing▪RunMap-reduce/Hiveeveryhour,every15min,every5min,…▪Howdowereduceper-batchoverhead?•StreamProcessing▪Aggregatethedataassoonasitarrives▪Howtosolvethereliabilityproblem?Decisions•StreamProcessingwins!•DataFreeway▪ScalableDataStreamFramework•Puma▪ReliableStreamAggregationEngineDataFreewayscalabledatastreamScribe•Simplepush/RPC-basedloggingsystem•Open-sourcedin2008.100logcategoriesatthattime.•Routingdrivenbystaticconfiguration.ScribeClientsScribeMid-TierScribeWritersNFSHDFSLogConsumerBatchCopiertail/fopen•9GB/secatpeak,10seclatency,2500logcategoriesDataFreewayScribeClientsCalligraphusMid-tierCalligraphusWritersHDFSHDFSC1C1C2C2DataNodeDataNodePTailZookeeperLogConsumerContinuousCopierPTail(intheplan)Calligraphus•RPCFileSystem▪Eachlogcategoryisrepresentedby1ormoreFSdirectories▪Eachdirectoryisanorderedlistoffiles•Bucketingsupport▪Applicationbucketsareapplication-definedshards.▪InfrastructurebucketsallowslogstreamsfromxB/stoxGB/s•Performance▪Latency:Callsyncevery7seconds▪Throughput:Easilysaturate1GbitNICContinuousCopier•FileSystemFileSystem•Lowlatencyandsmoothnetworkusage•Deployment▪Implementedaslong-runningmap-onlyjob▪Canmovetoanysimplejobscheduler•Coordination▪UselockfilesonHDFSfornow▪PlantomovetoZookeeperPTail•FileSystemStream(RPC)•Reliability▪Checkpointsinsertedintothedatastream▪Canrollbacktotailfromanydatacheckpoints▪Nodataloss/duplicatesdirectoryfilesdirectorydirectorycheckpointChannelComparisonPush/RPCPull/FSLatency1-2sec10secLoss/DupsFewNoneRobustnessLowHighComplexityLowHighPush/RPCPull/FSScribeCalligraphusPTail+ScribeSendContinuousCopierPumareal-timeaggregation/storageOverview•~1Mloglinespersecond,butlightread•MultipleGroup-Byoperationsperlogline•ThefirstkeyinGroupByisalwaystime/date-related•Complexaggregations:Uniqueusercount,mostfrequentelementsLogStreamAggregationsStorageServingMySQLandHBase:onepageMySQLHBaseParallelManualshardingAutomaticloadbalancingFail-overManualmaster/slaveswitchAutomaticReadefficiencyHighLowWriteefficiencyMediumHighColumnarsupportNoYesPuma2Architecture•PTailprovideparalleldatastreams•Foreachlogline,Puma2issue“increment”operationstoHBase.Puma2issymmetric(nosharding).•HBase:singleincrementonmultiplecolumnsPTailPuma2HBaseServingPuma2:ProsandCons•Pros▪Puma2codeisverysimple.▪Puma2serviceisveryeasytomaintain.•Cons▪“Increment”operationisexpensive.▪Donotsupportcomplexaggregations.▪Hackyimplementationof“mostfrequentelements”.▪Cancausesmalldataduplicates.ImprovementsinPuma2•Puma2▪Batchingofrequests.Didn’tworkwellbecauseoflong-taildistribution.•HBase▪“Increment”operationoptimizedbyreducinglocks.▪HBaseregion/HDFSfilelocality;short-circuitedread.▪Reliabilityimprovementsunderhighload.•Stillnotgoodenough!Puma3Architecture•Puma3isshardedbyaggregationkey.•Eachshardisahashmapinmemory.•Eachentryinhashmapisapairofanaggregationkeyandauser-definedaggregation.•HBaseaspersistentkey-valuestorage.PTailPuma3HBaseServingPuma3Architecture•Writeworkflow▪Foreachlogline,extractthecolumnsforkeyandvalue.▪Lookupinthehashmapandcalluser-definedaggregationPTailPuma3HBaseServingPuma3Architecture•Checkpointworkflow▪Every5min,savemodifiedhashmapentries,PTailcheckpointtoHBase▪Onstartup(afternodefailure),loadfromHBase▪GetridofitemsinmemoryoncethetimewindowhaspassedPTailPuma3HBaseServingPuma3Architecture•Readworkflow▪Readuncommitted:directlyservefromthein-memoryhashmap;loadfromHbaseonmiss.▪Readcommitted:readfromHBaseandserve.PTailPuma3HBaseServingPuma3Architecture•Join▪StaticjointableinHBase.▪Distributedhashlookupinuser-definedfunction(udf).▪Localcacheimprovesthethroughputoftheudfalot.PTailPuma3HBaseServingPuma2/Puma3comparison•Puma3ismuchbetterinwritethroughput▪Use25%oftheboxestohandlethesameload.▪HBaseisreallygoodatwritethroughput.•Puma3needsalotofmemory▪Use60GBofmemoryperboxforthehashmap▪SSDcanscaleto10xperbox.Puma3SpecialAggregations•UniqueCountsCalculation▪Adaptivesampling▪Bloomfilter(intheplan)•Mostfrequentitem(intheplan)▪Lossycounting▪ProbabilisticlossycountingPQL–PumaQueryLanguage•CREATEINPUTTABLEt(‘time',‘adid’,‘userid’);•CREATEVIEWvASSELECT*,udf.age(userid)FROMtWHEREudf.age(userid)21•CREATEHBASETABLEh…•CREATELOGICALTABLEl…•CREATEAGGREGATION‘abc’INSERTINTOl(a,b,c)SELECTudf.hour(time),adid,age,count(1),udf.count_distinc(userid)FROMvGROUPBYudf.hour(time),adid,age;FutureWorkschallengesandopportunitiesFutureWorks•SchedulerSupport▪Justneedsimpleschedulingbecausetheworkloadiscontinuous•Massadoption▪Migratemostdailyreportingqueriesfro