Spark-Streaming高級實踐課件

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

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

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

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