李遠(yuǎn)策2016-4-222016-4-22360Spark平臺介紹業(yè)務(wù)及應(yīng)用案例360-Spark集群概況解決Yarn和公司HDFS兼容解決Yarn和公司HDFS兼容問題，從standalone遷徙到Y(jié)arn機(jī)器學(xué)習(xí)集群200+節(jié)點(diǎn)1500+節(jié)點(diǎn)上線spark-1.6.0版本解決Yarn線上擴(kuò)容問題機(jī)器學(xué)習(xí)集群500+節(jié)點(diǎn)通用計(jì)算集群3000+節(jié)點(diǎn)每天作業(yè)數(shù)總和2W+支持幾十個業(yè)務(wù)部門使用上線spark-1.3.1版本standalone部署模式單個集群<100節(jié)點(diǎn)2015.052015.082015.112016.02上上線spark-1.4.1版本standalone+自研多租戶資源隔離機(jī)器學(xué)習(xí)集群<100節(jié)點(diǎn)節(jié)點(diǎn)360Spark平臺介紹360-Spark應(yīng)用MLLib?業(yè)務(wù)：新聞主題分類、新聞推薦、APP推薦、惡意代碼識別、惡意域名檢測等。GraphXPA?業(yè)務(wù)：搜索PageValue、網(wǎng)站安全監(jiān)測等。SparkSQL?采用HiveContext替換公司90%以上的Hive作業(yè)，每天例行1.5W+作業(yè)。?每個HiveSQL平均3輪MR作業(yè)，平均性能提升2~5倍。BDCRankonSparkBDCAASSRankonSpark內(nèi)置org.apache.spark.graphx.lib.PageRank算法不適合GraphX在千億邊規(guī)模上難以勝任PageRankonSpark(續(xù))vallinks=…//網(wǎng)頁傳遞概率RDD[(pageId,[(pageId,ratio)])]valranks=…//網(wǎng)頁valueRDD[(pageId,rank)]valratios=…//網(wǎng)頁賦值權(quán)重RDD[(pageId,ratio)]valtotalZ=…//網(wǎng)頁權(quán)重總和Longvalalpha=…//網(wǎng)也權(quán)重轉(zhuǎn)義比例floatvalcontribs=links.join(ranks).values.flatMap{case(urls,rank)=>urls.map(url=>(url._1,url._2*rank))}valranksTmp=contribs.reduceByKey(_+_).mapValues(alpha*_)valcontribsZ=ranksTmp.map(x=>x._2).sum()valsuperZ=totalZ-contribsZvalranksZ=ratios.map{x=>(x._1,x._2*superZ)}ranks=ranksTmp.rightOuterJoin(ranksZ).map{case(pageId,(rankTmp,rankZ))=>(pageId,rankTmp.getOrElse(0.0)+rankZ)}}12AFABDGC12AFABDGC33BBEEHHSpark做社群發(fā)現(xiàn)的歷程(1)lectionmutableMapSpark做社群發(fā)現(xiàn)的歷程(2)內(nèi)置org.apache.spark.graphx.lib.LabelPropagation經(jīng)常出現(xiàn)超大社區(qū)的問題，效果不太理想。Spark做社群發(fā)現(xiàn)的歷程(3)3、分層結(jié)果保存4、增量計(jì)算SparkSQL替換HiveHive遷移到SparkSQL的“正確打開方式”：1、編譯Spark加上-Phive-Phive-thriftserver參數(shù)2、部署Spark(Yarn)集群3、配置SparkSQL共用Hive的元數(shù)據(jù)庫4、用spark-hive(spark-sql)工具替換原有的hive命令5、-e/–f或者thriftserver提交作業(yè)。QL新版數(shù)據(jù)倉庫方案QL新版數(shù)據(jù)倉庫方案之之前數(shù)據(jù)倉庫方案Hive遷移SparkSQL–遷移方法SparkSQL版本選擇VSSparkSQLoin默認(rèn)HashJoin，支持BroadcastHashJoin和SortMergeJoin支持BroadcastHashOuterJoin支持的Hivemetastore版本0.12/0.13/1.0/1.1/1.2spark.sql.hive.metastore.versionspark.sql.hive.metastore.jars/path/your/hiveJars支持多HiveMetaStore版本實(shí)現(xiàn)原理1、HiveContext中創(chuàng)建一個URLClassLoader作為clientLoader加載不同版本的Hivemetastoreclasses。2、在ClientWrapper中調(diào)用Hive的接口時會切換到clientLoader中，然后借助HiveShim去反射Hive中對應(yīng)函數(shù)。例如360Spark平臺介紹業(yè)務(wù)及應(yīng)用案例經(jīng)驗(yàn)&改進(jìn)分享?SQL兼容(Insertoverwrite[local]directory的支持)因?yàn)镾parkSQL-HiveContext的SQL解析調(diào)用了Hive的ParseDriver.parse完成，所以語法解析上不存在問題。?SQL兼容(Insertoverwrite[local]directory的支持)1、解析AST中的TOK_DIR和TOK_LOCAL_DIR將其轉(zhuǎn)化成新定義的邏輯計(jì)劃WriteToDirectory2、將邏輯計(jì)劃WriteToDirectory轉(zhuǎn)換成新定義的物理計(jì)劃WriteToDirectory。3、在物理計(jì)劃WriteToDirectory執(zhí)行方法中復(fù)用InsertIntoHiveTable中的saveAsHiveFile邏輯將結(jié)果寫到4、如果是localdirectory則將結(jié)果再拉回到本地?SQL兼容(SQL二義性問題)selectC.idfrom(selectA.idfromtestbasAjoin(selectidfromtestb)BonA.id=B.id)C;C.idisA.idorB.id??transformationbugs(行尾部空列導(dǎo)致的數(shù)組越界)001\tABC\t002\t003\tEFG\t\t[001,ABC,002]GnewGenericInternalRow(prevLine.split(ioschema.outputRowFormatMap(“TOK_TABLEROWFORMATFIELD”)).map(CatalystTypeConverters.convertToCatalyst))讀線程寫線程outputStreamerrorStreaminputStream讀線程寫線程outputStreamerrorStreaminputStream?transformationbugs(Script的標(biāo)準(zhǔn)錯誤緩沖區(qū)打滿導(dǎo)致transform流程卡住)主主進(jìn)程ScriptScript進(jìn)程?輸入小文件合并的改進(jìn)(增加支持自定義inputFormat類)默認(rèn)采用建表時指定的InpurFormat，如果是默認(rèn)的TextInputFormat，當(dāng)小文件比較多是可能會導(dǎo)致RDD的partition數(shù)太多，導(dǎo)致性能下降。解決辦法：通過參數(shù)允許用戶指定InputFormat，在TableReader中反射生成對應(yīng)的InputFormat對象并傳入到HadoopRDD的構(gòu)造函數(shù)中。使用方法：setspark.sql.hive.inputformat=org.apache.hadoop.mapred.lib.CombineTextInputFormat;?輸出小文件合并的改進(jìn)(增加自動合并結(jié)果文件)當(dāng)spark.sql.shuffle.partitions設(shè)置的比較大且結(jié)果數(shù)據(jù)集比較小時，會產(chǎn)生大量的小文件(文件數(shù)等同spark.sql.shuffle.partitions)。解決辦法：在最后的執(zhí)行計(jì)劃中加入一個repartitiontransformation。通過參數(shù)控制最終的partitions數(shù)且不影響shufflepartition的數(shù)量。?支持yarn-cluster模式，減小client的負(fù)載默認(rèn)的yarn-client模式下Scheduler會運(yùn)行在client上，加重client機(jī)器的負(fù)載。解決辦法：讓sparkSQL工具支持yarn-cluster模式。1)在Yarn集群上部署SparkSQL依賴的hivemetastorejar包。2)開通Yarnnodemanager節(jié)點(diǎn)訪問Hivemetastore數(shù)據(jù)庫3)解決“\”“轉(zhuǎn)義問題。如spark-hive–e“select*fromuserwherename=\”張三””;在yarn-cluster模式中會觸發(fā)兩commandSpark-1.4.1HashJoinJoin數(shù)據(jù)傾斜rtitionvalue=Avalue=Bvalue=Cvalue=Dvalue=Evalue=Frtitionvalue=Avalue=Bvalue=Cvalue=Dvalue=Evalue=FJoin數(shù)據(jù)傾斜rtitionvalue=Avalue=Bvalue=Cvalue=Dvalue=Evalue=Frtitionvalue=Avalue=Bvalue=Cvalue=Dvalue=Evalue=FSpark-1.6.0SortMergeJo