Spark-Streaming高級(jí)實(shí)踐課件

SparkStreaming高級(jí)實(shí)踐王富平2018年3月Spark-Streaming簡(jiǎn)介1SPARK是什么？批處理框架?MapReduce的進(jìn)化版高性能?內(nèi)存加速、分布式計(jì)算生態(tài)豐富機(jī)器學(xué)習(xí)庫(kù)(mlib)、圖計(jì)算(graghX)、Sql、SparkStreamingSpark-Streaming簡(jiǎn)介1SPARKStreaming是什么？實(shí)時(shí)處理框架?基于批處理理念設(shè)計(jì)的實(shí)時(shí)處理框架高性能?內(nèi)存加速、分布式計(jì)算生態(tài)豐富機(jī)器學(xué)習(xí)庫(kù)(mlib)、圖計(jì)算(graghX)、SqlSpark-Streaming簡(jiǎn)介1基于批處理理念設(shè)計(jì)的實(shí)時(shí)處理框架基于spark先天優(yōu)勢(shì)，與各類型數(shù)據(jù)庫(kù)、數(shù)據(jù)源結(jié)合較好朋友圈強(qiáng)大，spark的sql、mlib等組件，在sparkstreaming可以直接試用Spark-Streaming簡(jiǎn)介1WordCount例子Spark-Streaming高級(jí)特性—Window特性2WordCount需求升級(jí)：每10s統(tǒng)計(jì)一次單詞在前1分鐘內(nèi)出現(xiàn)次數(shù)需求特點(diǎn)：數(shù)據(jù)復(fù)用?當(dāng)前批次計(jì)算需要使用前一批次數(shù)據(jù)。聚合操作?詞頻計(jì)算，是對(duì)前一分鐘內(nèi)數(shù)據(jù)進(jìn)行聚合操作Spark-Streaming高級(jí)特性—Window特性2Windowwidth長(zhǎng)度該設(shè)置為多少？Slidinginternal該如何設(shè)置？Batchinternal該如何設(shè)置？BatchInternal?分批時(shí)間間隔Windowwidth?窗口長(zhǎng)度?窗口長(zhǎng)度必須是”batchinternal”整數(shù)倍SlidingInternal?窗口步進(jìn)間隔?窗口步進(jìn)間隔必須是”batchinternal”整數(shù)倍Spark-Streaming高級(jí)特性—Window特性2函數(shù)名參數(shù)描述windowwindowDuration：窗口長(zhǎng)度創(chuàng)建窗口計(jì)算流windowwindowDuration：窗口長(zhǎng)度slideDuration：滑動(dòng)間隔創(chuàng)建窗口計(jì)算流創(chuàng)建WINDOW數(shù)據(jù)流Spark-Streaming高級(jí)特性—Window特性2函數(shù)名參數(shù)描述countByWindowwindowDuration：窗口長(zhǎng)度slideDuration：滑動(dòng)間隔返回窗口內(nèi)記錄數(shù)countByValueAndWindowwindowDuration：窗口長(zhǎng)度slideDuration：滑動(dòng)間隔返回窗口內(nèi)不同value的記錄數(shù)WINDOW相關(guān)計(jì)算函數(shù)Spark-Streaming高級(jí)特性—Window特性2計(jì)算窗口期內(nèi)：請(qǐng)求失敗率Spark-Streaming高級(jí)特性—Sql特性3Sparkstream數(shù)據(jù)封裝成DataFrame（DataSet）之后，天然可以使用sparksql特性注冊(cè)臨時(shí)表valrecordsDF=spark.createDataFrame((1to100).map(i=>Record(i,s"val_$i")))recordsDF.createOrReplaceTempView("records")sql("SELECT*FROMrecordsrJOINsrcsONr.key=s.key").show()表joinsql("SELECT*FROMrecordsrJOINsrcsONr.key=s.key").show()Spark-Streaming高級(jí)特性—Sql特性3定義UDF加載jar包udfvalsqlContext=newHiveContext(sc)sqlContext.sql(“createtemporaryfunctiongetCateas'com.yihaodian.udf.hive.GetUrlCate'USINGJAR'SearchUDF-1.1-SNAPSHOT.jar‘”)動(dòng)態(tài)定義UDFspark.udf.register("upper",(in:String)=>in.toUpperCase)Spark-Streaming高級(jí)特性—checkpoint機(jī)制4Stream通過(guò)使用CheckPoint保存處理狀態(tài)甚至當(dāng)前處理數(shù)據(jù)，一旦任務(wù)失敗后，可以利用checkPoint數(shù)據(jù)進(jìn)行恢復(fù)checkpoint特點(diǎn)：Metadatacheckpointing?保存stream任務(wù)配置?保存stream處理邏輯?保存stream當(dāng)前未完成數(shù)據(jù)處理任務(wù)Datacheckpointing?保存inputstream輸入數(shù)據(jù)，當(dāng)任務(wù)異常退出恢復(fù)后，數(shù)據(jù)重新處理?保存狀態(tài)數(shù)據(jù)，比如reduceByWindow、updateStateByKey需要使用上一批次的數(shù)據(jù)Spark-Streaming高級(jí)特性—checkpoint機(jī)制3sparkstreaming從1.2開(kāi)始提供了數(shù)據(jù)的零丟失，想享受這個(gè)特性，需要滿足如下條件?可靠的sources和可靠的receivers?應(yīng)用metadatacheckpoint?WAL（writeaheadlog）也就是datacheckpoint無(wú)WALSpark-Streaming高級(jí)特性—checkpoint機(jī)制4加入WALSpark-Streaming高級(jí)特性—checkpoint機(jī)制3Wal缺點(diǎn)?降低了receivers的性能，因?yàn)閿?shù)據(jù)還要存儲(chǔ)到HDFS等分布式文件系統(tǒng)?只能保證At-Least-Once，不能保證exactly-onceKafkadirectAPI?為了WAL的性能損失和exactly-once，sparkstreaming1.3中使用KafkadirectAPI。非常巧妙，Sparkdriver計(jì)算下個(gè)batch的offsets，指導(dǎo)executor消費(fèi)對(duì)應(yīng)的topics和partitions。消費(fèi)Kafka消息，就像消費(fèi)文件系統(tǒng)文件一樣。NDCG指標(biāo)計(jì)算3NDCG是什么NormalizedDiscountedCumulativeGain，即NDCG，常用作搜索排序的評(píng)價(jià)指標(biāo)，理想情況下排序越靠前的搜索結(jié)果，點(diǎn)擊概率越大，即得分越高(gain)。CG=排序結(jié)果的得分求和，discounted是根據(jù)排名，對(duì)每個(gè)結(jié)果得分*排名權(quán)重，權(quán)重=1/log(1+排名)，排名越靠前的權(quán)重越高。首先我們計(jì)算理想DCG（稱之為IDCG），再根據(jù)用戶點(diǎn)擊結(jié)果，計(jì)算真實(shí)的DCG，NDCG=DCG/IDCG，值越接近1，則代表搜索結(jié)果越好。DCG計(jì)算公式如下NDCG指標(biāo)計(jì)算4在蘇寧易購(gòu)搜索關(guān)鍵詞“蘋果”，取第一排4個(gè)結(jié)果為例子NDCG指標(biāo)計(jì)算4

結(jié)果1結(jié)果2結(jié)果3結(jié)果4實(shí)際點(diǎn)擊次數(shù)0010理想點(diǎn)擊次數(shù)1000IDCGDCG點(diǎn)擊結(jié)果計(jì)算得出IDCG=1，DCG=0.5，NDCG=DCG/IDCG=0.5NDCG在sparkstreaming實(shí)現(xiàn)4數(shù)據(jù)調(diào)研86%的搜索行為在5分鐘內(nèi)完成90%的搜索行為在10分鐘內(nèi)完成時(shí)間窗口計(jì)算?對(duì)前15分鐘數(shù)據(jù)的整體分析去重?一次搜索只計(jì)算一次NDCG在sparkstreaming實(shí)現(xiàn)4試用SparkStreaming框架的Window特性，實(shí)現(xiàn)時(shí)間窗口計(jì)算。時(shí)間窗口為15分鐘，步進(jìn)5分鐘，意味著每5分鐘計(jì)算一次每次計(jì)算，只對(duì)在區(qū)間［15分鐘前，10分鐘前］發(fā)起的搜索行為進(jìn)行NDCG計(jì)算，這樣就不會(huì)造成重復(fù)計(jì)算NDCG性能保障—容量預(yù)估5容量預(yù)估不是一個(gè)靜態(tài)工作流量日志在不斷增長(zhǎng)，而系統(tǒng)處理能力是有限的

大促活動(dòng)會(huì)造成額外的數(shù)據(jù)高峰。NDCG性能保障—容量預(yù)估5容量調(diào)節(jié)KafkaTopic分區(qū)數(shù)量SparkStreaming并發(fā)數(shù)SparkStreaming

cpu、內(nèi)存NDCG性能保障—多維計(jì)算優(yōu)化6NDCG指標(biāo)為例子，目前支持4個(gè)維度組合的計(jì)算：大區(qū)、城市、渠道、搜索詞，為了支持4個(gè)維度任意組合，需要進(jìn)行15次計(jì)算，在HBas