SparkStreaming

Large-scalenear-real-timestreamprocessing

TathagataDas(TD)UCBerkeleyUCBERKELEYWhatisSparkStreaming?FrameworkforlargescalestreamprocessingScalesto100sofnodesCanachievesecondscalelatenciesIntegrateswithSpark’sbatchandinteractiveprocessingProvidesasimplebatch-likeAPIforimplementingcomplexalgorithmCanabsorblivedatastreamsfromKafka,Flume,ZeroMQ,etc.

MotivationManyimportantapplicationsmustprocesslargestreamsoflivedataandprovideresultsinnear-real-timeSocialnetworktrendsWebsitestatisticsIntrustiondetectionsystemsetc.RequirelargeclusterstohandleworkloadsRequirelatenciesoffewsecondsNeedforaframework……forbuildingsuchcomplexstreamprocessingapplicationsButwhataretherequirementsfromsuchaframework?RequirementsScalable

tolargeclustersSecond-scalelatenciesSimple

programmingmodelCasestudy:Conviva,Inc.Real-timemonitoringofonlinevideometadataHBO,ESPN,ABC,SyFy,…TwoprocessingstacksCustom-builtdistributedstreamprocessingsystem1000scomplexmetricsonmillionsofvideosessionsRequiresmanydozensofnodesforprocessingHadoopbackendforofflineanalysisGeneratingdailyandmonthlyreportsSimilarcomputationasthestreamingsystemCustom-builtdistributedstreamprocessingsystem1000scomplexmetricsonmillionsofvideossessionsRequiresmanydozensofnodesforprocessingHadoopbackendforofflineanalysisGeneratingdailyandmonthlyreportsSimilarcomputationasthestreamingsystemCasestudy:XYZ,Inc.AnycompanywhowantstoprocesslivestreamingdatahasthisproblemTwicetheefforttoimplementanynewfunctionTwicethenumberofbugstosolveTwicetheheadacheTwoprocessingstacksRequirementsScalable

tolargeclustersSecond-scalelatenciesSimple

programmingmodelIntegrated

withbatch&interactiveprocessingStatefulStreamProcessingTraditionalstreamingsystemshaveaevent-drivenrecord-at-a-timeprocessingmodelEachnodehasmutablestateForeachrecord,updatestate&sendnewrecordsStateislostifnodedies!Makingstatefulstreamprocessingbefault-tolerantischallengingmutablestatenode1node3inputrecordsnode2inputrecords9ExistingStreamingSystemsStormReplaysrecordifnotprocessedbyanodeProcesseseachrecordatleastonceMayupdatemutablestatetwice!Mutablestatecanbelostduetofailure!Trident–UsetransactionstoupdatestateProcesseseachrecordexactlyoncePerstatetransactionupdatesslow10RequirementsScalable

tolargeclustersSecond-scalelatenciesSimple

programmingmodelIntegrated

withbatch&interactiveprocessingEfficientfault-tolerance

instatefulcomputationsSparkStreaming12DiscretizedStreamProcessingRunastreamingcomputationasaseriesofverysmall,deterministicbatchjobs13SparkSparkStreamingbatchesofXsecondslivedatastreamprocessedresultsChopupthelivestreamintobatchesofXsecondsSparktreatseachbatchofdataasRDDsandprocessesthemusingRDDoperationsFinally,theprocessedresultsoftheRDDoperationsarereturnedinbatchesDiscretizedStreamProcessingRunastreamingcomputationasaseriesofverysmall,deterministicbatchjobs14SparkSparkStreamingbatchesofXsecondslivedatastreamprocessedresultsBatchsizesaslowas?second,latency~1secondPotentialforcombiningbatchprocessingandstreamingprocessinginthesamesystemExample1–GethashtagsfromTwitterval

tweets

=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)DStream:asequenceofRDDrepresentingastreamofdatabatch@t+1batch@tbatch@t+2tweetsDStreamstoredinmemoryasanRDD(immutable,distributed)TwitterStreamingAPIExample1–GethashtagsfromTwittervaltweets=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)val

hashTags

=tweets.flatMap

(status=>getTags(status))flatMapflatMapflatMap…transformation:modifydatainoneDstreamtocreateanotherDStream

newDStreamnewRDDscreatedforeverybatchbatch@t+1batch@tbatch@t+2tweetsDStreamhashTags

Dstream[#cat,#dog,…]Example1–GethashtagsfromTwittervaltweets=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)val

hashTags=tweets.flatMap(status=>getTags(status))hashTags.saveAsHadoopFiles("hdfs://...")outputoperation:topushdatatoexternalstorageflatMapflatMapflatMapsavesavesavebatch@t+1batch@tbatch@t+2tweetsDStreamhashTagsDStreameverybatchsavedtoHDFSJavaExampleScalaval

tweets

=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)val

hashTags

=tweets.flatMap

(status=>getTags(status))hashTags.saveAsHadoopFiles("hdfs://...")JavaJavaDStream<Status>tweets=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)JavaDstream<String>hashTags

=tweets.flatMap(newFunction<...>{})hashTags.saveAsHadoopFiles("hdfs://...")FunctionobjecttodefinethetransformationFault-toleranceRDDsarerememberthesequenceofoperationsthatcreateditfromtheoriginalfault-tolerantinputdataBatchesofinputdataarereplicatedinmemoryofmultipleworkernodes,thereforefault-tolerantDatalostduetoworkerfailure,canberecomputedfrominputdatainputdatareplicatedinmemoryflatMaplostpartitionsrecomputedonotherworkerstweetsRDDhashTagsRDDKeyconceptsDStream–sequenceofRDDsrepresentingastreamofdataTwitter,HDFS,Kafka,Flume,ZeroMQ,AkkaActor,TCPsocketsTransformations–modifydatafromonDStreamtoanotherStandardRDDoperations–map,countByValue,reduce,join,…Statefuloperations–window,countByValueAndWindow,…OutputOperations–senddatatoexternalentitysaveAsHadoopFiles–savestoHDFSforeach–doanythingwitheachbatchofresultsExample2–Countthehashtagsvaltweets=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)val

hashTags=tweets.flatMap(status=>getTags(status))val

tagCounts=hashTags.countByValue()flatMapmapreduceByKeyflatMapmapreduceByKey…flatMapmapreduceByKeybatch@t+1batch@tbatch@t+2hashTagstweetstagCounts[(#cat,10),(#dog,25),...]Example3–Countthehashtagsoverlast10minsvaltweets=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)val

hashTags=tweets.flatMap(status=>getTags(status))val

tagCounts=hashTags.window(Minutes(10),Seconds(1)).countByValue()slidingwindowoperationwindowlengthslidingintervaltagCountsExample3–Countingthehashtagsoverlast10minsval

tagCounts

=hashTags.window(Minutes(10),Seconds(1)).countByValue()hashTagst-1tt+1t+2t+3slidingwindowcountByValuecountoverallthedatainthewindow?Smartwindow-basedcountByValueval

tagCounts=hashtags.countByValueAndWindow(Minutes(10),Seconds(1))

hashTagst-1tt+1t+2t+3++–countByValueaddthecountsfromthenewbatchinthewindowsubtractthecountsfrombatchbeforethewindowtagCountsSmartwindow-basedreduceTechniquetoincrementallycomputecountgeneralizestomanyreduceoperationsNeedafunctionto“inversereduce”(“subtract”forcounting)Couldhaveimplementedcountingas:

hashTags.reduceByKeyAndWindow(_+_,_-_,Minutes(1),…)25DemoFault-tolerantStatefulProcessingAllintermediatedataareRDDs,hencecanberecomputediflost

hashTagst-1tt+1t+2t+3tagCountsFault-tolerantStatefulProcessingStatedatanotlostevenifaworkernodediesDoesnotchangethevalueofyourresultExactlyoncesemanticstoalltransformationsNodoublecounting!28OtherInterestingOperationsMaintainingarbitrarystate,tracksessionsMaintainper-usermoodasstate,andupdateitwithhis/hertweets

tweets.updateStateByKey(tweet=>updateMood(tweet))DoarbitrarySparkRDDcomputationwithinDStreamJoinincomingtweetswithaspamfiletofilteroutbadtweets

tweets.transform(tweetsRDD=>{

tweetsRDD.join(spamHDFSFile).filter(...)})PerformanceCanprocess6GB/sec(60Mrecords/sec)ofdataon100nodesatsub-secondlatencyTestedwith100streamsofdataon100EC2instanceswith4coreseach30ComparisonwithStormandS4HigherthroughputthanStormSparkStreaming:670krecords/second/nodeStorm:115krecords/second/nodeApacheS4:7.5krecords/second/node31FastFaultRecoveryRecoversfromfaults/stragglerswithin1sec32RealApplications:ConvivaReal-timemonitoringofvideometadata33Achieved1-2secondlatencyMillionsofvideosessionsprocessedScaleslinearlywithclustersizeRealApplications:MobileMillenniumProjectTraffictransittimeestimationusingonlinemachinelearningonGPSobservations34MarkovchainMonteCarlosimulationsonGPSobservationsVeryCPUintensive,requiresdozensofmachinesforusefulcomputationScaleslinearlywithclustersizeVision-onestacktorulethemallAd-hocQueriesBatchProcessingStreamProcessingSpark+Shark+SparkStreamingSparkprogramvsSparkStreamingprogramSparkStreamingprogramonTwitterstreamval

tweets

=ssc.twitterStream(<Twitterusername>,<Twitterpassword>)val

hashTags

=tweets.flatMap

(status=>getTags(status))hashTags.saveAsHadoopFiles("hdfs://...")SparkprogramonTwitterlogfileval

tweets

=sc.hadoopFile("hdfs://...")val

hashTags

=tweets.flatMap

(status=>getTags(status))hashTags.saveAsHadoopFile("hdfs://...")Vision-onestacktorulethemallExploredatainteractivelyusingSparkShell/PySparktoidentifyproblemsUsesamecodeinSparkstand-aloneprogramstoidentifyproblemsinproductionlogsUsesimilarcodeinSparkStreamingtoidentifyproblemsinlivelogstreams$./spark-shellscala>valfile=sc.hadoopFile(“smallLogs”)...scala>valfiltered=file.filter(_.contains(“ERROR”))...scala>valmapped=file.map(...)...objectProcessProductionData{

defmain(args:Array[String]){

val

sc=newSparkContext(...)

valfile=sc.hadoopFile(“productionLogs”)

valfiltered=file.filter(_.contains(“ERROR”))

valmapped=file.map(...)...}}objectProcessLiveStream{

defmain(args:Array[String]){

val

sc=newStreamingContext(...)

valstream=sc.kafkaStream(...)

valfiltered=file.filter(_.contains(“ERROR”))

valmapped=file.map(...)...}}Vision-onestacktorulethemallExploredatainteractivelyusingSparkShell/PySparktoidentifyproblemsUsesamecodeinSparkstand-aloneprogramstoidentifyproblemsinproductionlogsUsesimilarcodeinSparkStreamingtoidentifyproblemsinlivelogstreams$./spark-shellscala>valfile=sc.hadoopFile(“smallLogs”)...scala>valfiltered=