崔波-京東商城-數(shù)據(jù)挖掘工程師-SparkStreaming在京東廣告跟蹤系統(tǒng)里的應(yīng)用

ng?1.需求與挑戰(zhàn)與架構(gòu)對比?2.技術(shù)版本選型?3.sparkstreaming的優(yōu)缺點(diǎn)?4.設(shè)計(jì)取舍與最佳實(shí)踐?1.需求與挑戰(zhàn)與架構(gòu)對比?2.技術(shù)版本選型?3.sparkstreaming的優(yōu)缺點(diǎn)?4.設(shè)計(jì)取舍與最佳實(shí)踐?評估廣告主投放效果?分鐘級延遲?利于廣告主實(shí)時(shí)優(yōu)化?大促的性能需求?所依賴用戶行為日志巨量?需要若干天的廣告點(diǎn)擊等?之前重度依賴?技術(shù)棧+數(shù)據(jù)存儲棧分化?離線與實(shí)時(shí)處理融合?Datapipline的一致性?可靠的大容量流式處理系統(tǒng)?快速可擴(kuò)展?擴(kuò)容與遷移，監(jiān)控運(yùn)維?簡潔高效的scala代碼?良好的容錯?端對端的全容錯?spark結(jié)合kafka＝>exactlyonce的數(shù)據(jù)處理?SparkRDD=>按照lineage關(guān)系重做自適應(yīng)性?技術(shù)棧?結(jié)合實(shí)時(shí)與離線業(yè)務(wù)系統(tǒng)概覽(新系統(tǒng))?新的效果追蹤系統(tǒng)-v2?1.需求與挑戰(zhàn)與架構(gòu)對比?2.技術(shù)版本選型?3.sparkstreaming的優(yōu)缺點(diǎn)?4.設(shè)計(jì)取舍與最佳實(shí)踐?業(yè)務(wù)特點(diǎn)每批次遍歷海量數(shù)據(jù)，有復(fù)雜join跟sort?沿用經(jīng)典的streaming編程api?Structurestreaming抽象度更高?dataset作為主要數(shù)據(jù)建模接口?重使用transform/transformwith：?涉及雙流合并問題?同時(shí)會使用離線數(shù)據(jù)?性能跟穩(wěn)定性更好的版本Sparkstandalone模式40node4C*64G內(nèi)存?Spark2.0.1?新的tungsten引擎(高端db?全階段代碼優(yōu)化?Cache感知算法?更好的parquet并行解析器1.非虛函數(shù)調(diào)用寄存器級優(yōu)化更好的字節(jié)碼優(yōu)化太多虛函數(shù)調(diào)用cache不友好無法深入的編譯級優(yōu)化?1.需求與挑戰(zhàn)與架構(gòu)對比?2.技術(shù)版本選型?3.sparkstreaming的優(yōu)缺點(diǎn)?4.設(shè)計(jì)取舍與最佳實(shí)踐?動態(tài)可擴(kuò)展跟自適應(yīng)性?易于編程:dataSet,dataFrame,Sql,RDD?StreamingAPI可以很容易的跟batchAPI融合backpressureenabledtrueafkamaxRatePerPartitiontion負(fù)載流量自動調(diào)整資源dynamicAllocationenabledtrueAllocationmaxExecutors?程序哪部分執(zhí)行一次，哪部分每個(gè)批次都執(zhí)行？eamsAPITransform/transformwithAPI一般都是RDD，本質(zhì)是多批次的RDD。行會產(chǎn)生大量的碎片化的RDD。的數(shù)據(jù)比較難Slice依賴于時(shí)間?過度設(shè)計(jì)?過高的抽象度導(dǎo)致對一些簡單的操作很難?checkpoint:跟數(shù)據(jù)升級后的優(yōu)雅上線都有?1.需求與挑戰(zhàn)與架構(gòu)對比?2.技術(shù)版本選型?3.sparkstreaming的優(yōu)缺點(diǎn)?4.設(shè)計(jì)取舍與最佳實(shí)踐?調(diào)整一些默認(rèn)的參數(shù)?不使用streaming本身的checkpoint功能oint?避免使用大窗口Dstream?不使用流控跟反壓功能壓導(dǎo)致數(shù)據(jù)不準(zhǔn)確?引入alluxio?AlluxioandHDFSfheapmemory?快速的離線數(shù)據(jù)每日熱加載?挑戰(zhàn)：運(yùn)維復(fù)雜度的增加，內(nèi)存監(jiān)控?更激進(jìn)的思路?放棄streaming，將輸入數(shù)據(jù)存入alluxio內(nèi)存，用化理代碼統(tǒng)一?適用于時(shí)間粒度較粗(>幾分鐘)，對啟動時(shí)間與性能不苛Dstreambatchdata?Dstream碎片?多天的window導(dǎo)致碎片膨脹。?只保存一天的window。?離線與實(shí)時(shí)數(shù)據(jù)整合?狀態(tài)系統(tǒng)觸發(fā)離線數(shù)據(jù)加載或替換(OnlinedStream)union(ETLedbatchcoalesce?自己做checkpoint(僅data)?程序保存kafka的offset?每次啟動讀取lastcommitoffset，每個(gè)批次結(jié)束offset做commit。?usecreateDirectStreamnotcreateDstream?更嚴(yán)格的exactlyonce需要commit本批次要處理的offsetrange要按照offset的區(qū)間映射唯一文件數(shù)據(jù)的持久化跑數(shù)據(jù)需求dataFrame更適用于ETL?不涉及復(fù)雜業(yè)務(wù)處理的環(huán)節(jié)，更高級的半程序化語言的sql?帶schema的行集合,2.0中：dataFrame＝dataSet[Row]dataSet適用于業(yè)務(wù)邏輯實(shí)現(xiàn)(fromspark1.6)?帶類型信息的dataFrame：?融合了RDD，dataFrame優(yōu)點(diǎn)，更優(yōu)雅的代碼?Bestpractice：盡可能使用dataSet代替dataFrame，不要再使用RDDWindowpartitionby的重寫?直接用在dataSet上性能會有問題?exampleDF.withColumn(“xx”,row_number()overWindow.partitionBy(col1,col2).orderBy(col3,col4desc)).filter($"rank"===1).withColumn…?exampleDS.repartition(200,col1,col2).sortWithinPartitions(col3,col4,desc).mapPartitions(iter=>Avoidshuffleandskewtition?窄依賴與shuffle?相同的key以及分片數(shù)?表Join順序in?broadcast變量瘦身?根據(jù)輸入量大小自動適配不同邏輯(避免Avoidshuffleandskew?盡量走mergeSortJoin?避免or條件以及復(fù)雜條件join?數(shù)據(jù)傾斜三板斧?考慮業(yè)務(wù)上過濾異常數(shù)據(jù)?是否可以用sort代替join？?m*n>m+n?Ds1.joinwith(ds2)=>?Ds1.union(ds2).repartiton(num,cols1).smapPartitions?考慮復(fù)雜的分片方案的jdbc功能?對批量插入優(yōu)化，細(xì)粒度控制不太方便?有異常處理的bug，且對dataset的支持不好?解決1：用scala的反射重新實(shí)現(xiàn)jdbc的封裝(后續(xù)可以提交給社區(qū))?解決2：或者用ORM框架直接支持寫入?Par