騰訊在Spark上的應(yīng)用與實(shí)踐優(yōu)化.pdf

騰訊在spark上的應(yīng)用與實(shí)踐優(yōu)化 王聯(lián)輝 agenda spark在騰訊的當(dāng)前現(xiàn)狀 spark在騰訊的典型應(yīng)用及效果 騰訊在spark上的實(shí)踐和優(yōu)化 未來(lái)工作和計(jì)劃 tdw(tencent distributed data warehouse) overview mapreduce spark hive/pig ide lhoste(workflow system) gaia(based on yarn) gaia集群結(jié)點(diǎn)數(shù): 8000+ hdfs的存儲(chǔ)空間: 150pb+ 每天新增數(shù)據(jù): 1pb+ 每天仸務(wù)數(shù): 1m+ 每天計(jì)算量: 10pb+ user defined jobs (scala/java/python/) hdfs storm docker ide: 用于提交sql戒腳本的eclipse插件和web界面 lhoste: 各類作業(yè)的工作流調(diào)度系統(tǒng)，類似于oozie gaia: 基于yarn進(jìn)行定制和優(yōu)化的資源管理系統(tǒng) user 為什么我們引入spark? dag計(jì)算引擎 data cache & sharing 其他: task調(diào)度，數(shù)據(jù)廣播等 m m m r r hdfs m m r hdfs m m r hdfs m m r m m m r r m m r r r hdfs iter.1 iter.2 iter.3 result1 result2 result3 hdfs iter.1 iter.2 iter.3 result1 result2 result3 spark在騰訊的當(dāng)前現(xiàn)狀 作業(yè)類型：etl,sparksql,machine learning,graph compute,streaming 每天仸務(wù)數(shù)：10k+ 部署模式：gaia(8000+ nodes,with 24 cores and 60g memory each) 底層存儲(chǔ): hdfs/hive/hbase/mysql/ 從2013年的spark 0.6版本開始，目前的版本是spark1.2 spark spark sql spark streaming mllib (machine learning) graphx scala/java/ python etl hdfs hive table hbase mysql postgresql others spark在騰訊的典型應(yīng)用及效果 case 1: 預(yù)測(cè)用戶的廣告點(diǎn)擊概率 case 2: 計(jì)算二個(gè)好友間的共同好友數(shù) case 3: 用于etl的sparksql和dag任務(wù) case 1: 預(yù)測(cè)用戶的廣告點(diǎn)擊概率 gender age marital status location ad. one ad. two is click? man 20 unmarried newyork true false yes woman 40 married california false true yes man 60 married california false true no gende r age marital status location ad. one ad. two clicks possibility man 30 unmarried newyork true false ? man 30 unmarried newyork false true ? gender age marital status location ad. one ad. two clicks possibility man 30 unmarried newyork true false 30% man 30 unmarried newyork false true 50% w man woman ad. 1 ad.2 -0.896 0.895 -0.490 0.489 model train at spark table-1 table-2 table-3 table-4 實(shí)時(shí)預(yù)處理訓(xùn)練數(shù)據(jù)并推送到hdfs上 將訓(xùn)練數(shù)據(jù)(200g+) 導(dǎo)入至rdd & cache 初始化隨機(jī)值給w并廣播至各個(gè)executor 迭代訓(xùn)練60次左史 for (i p.x * (1 / (1 + exp(-p.y * (w.dot(p.x) - 1) * p.y .reduce(_ + _) w -= gradient 將模型推送廣告后臺(tái)在線服務(wù)器 每個(gè)計(jì)算的時(shí)間在1015分鐘 hdfs driver executor cache task/rdd executor cache task/rdd task/rdd task/rdd task/rdd w w w = w-gradient gradient gradient data collect tools case 1: 預(yù)測(cè)用戶的廣告點(diǎn)擊概率 user friend 2 1 4 1 1 2 5 3 6 3 6 5 5 6 3 6 user friend number of mutual friends 2 1 0 4 1 0 1 2 0 5 3 1 6 3 0 6 5 1 5 6 1 3 6 0 5億用戶數(shù)，500億條邊 10億用戶數(shù)，1000億條邊 user links table (id,fid) user friends table (id,fids) group by temp-1 table (id,fid,id-fids) join by temp table (id,fid,id-fids,fid-fids) join by result table (id,fid,count) case 2: 計(jì)算二個(gè)好友間的共同好友數(shù) table-1 table-2 case 2: 計(jì)算二個(gè)好友間的共同好友數(shù) user friend 2 1 4 1 1 2 5 3 6 3 6 5 5 6 3 6 pid id fid 1 2 1 1 2 2 3 6 3 4 1 5 3 6 3 4 6 5 5 6 val partitionstrategy = partitionstrategy.edgepartition2d table-1 table-2 pid id fid 1 2 1 1 2 2 3 6 3 4 1 5 3 6 3 4 6 5 5 6 id pid-list 1 1,3 2 1 3 2,3 4 3 5 3,4 6 2,3,4 pid uid-friendlist 1 1-2 2 - 1 2 3-6 6 -3,5 3 1-2 3 -6 4 -1 5 -3,6 6-3,5 4 5 -3,6 6 -3,5 id friend-list 1 2 2 1 3 6 4 1 5 3,6 6 3,5 id pid- list friend- list 1 1,3 2 2 1 1 3 2,3 6 4 3 1 5 3,4 3,6 6 2,3,4 3,5 join by id join by pid user friend total number of mutual friends 2 1 0 4 1 0 case 2: 計(jì)算二個(gè)好友間的共同好友數(shù) table-1 table-2 table-3 table-4 table-5 table-6 edges/vertex nodes hive(map reduce) spark executors memory,cores/par titions 50b/0.5b 200 12 h 2 h 15g/1/5000 50b/0.5b 1000 - 43m 15g/1/5000 100b/1b 200 24 h 6h 25g/1/10000 100b/1b 1000 - 2h 25g/1/10000 one node, 55g memory,2*12t sata disk,24*1.9ghz cpu,1gbps network 根據(jù)shuffle的數(shù)據(jù)量來(lái)確定partition數(shù) 使用sort-based shuffle來(lái)提升性能和減少reduce的內(nèi)存使用(spark-1.2). 在大集群時(shí)當(dāng)連接超時(shí)后選擇重試來(lái)減少executor丟失的概率(spark-1.2). 為避免executor被yarn給kill,通常需要設(shè)置 spark.yarn.executor.memoryoverhead. case 2: 計(jì)算二個(gè)好友間的共同好友數(shù) case 3: 用于etl的sparksql和dag仸務(wù) insert table test_result select t3.d ,count(*) from( select distinct a,b from join_1) t1 join ( select distinct b ,c from join_2) t2 on (t1.a = t2.c) join ( select distinct c ,d from join_3) t3 on (t2.b = t3.d) group by t3.d compute resources time hive 200 maps, 50 reduces(1 core, 3.2g memory) 30 min sparksql 50 executors(1 core, 4g memory) 5 min 騰訊在spark上的實(shí)踐和優(yōu)化 應(yīng)用程序開發(fā)中的使用經(jīng)驗(yàn) 對(duì)于etl作業(yè)使用勱態(tài)資源擴(kuò)縮容特性 redcue階段在map階段未全部完成前執(zhí)行 基于數(shù)據(jù)的大小預(yù)測(cè)stage的partition數(shù) 為sparksql的每個(gè)session分配一個(gè)driver count(distinct)的優(yōu)化 基于排序的groupby/join 應(yīng)用程序開發(fā)中的使用經(jīng)驗(yàn) 當(dāng)有小表時(shí)使用broadcast join代替common join 盡量使用reducebykey來(lái)代替groupbykey 設(shè)置spark.serializer=org.apache.spark.serializer.kryoserializer 根據(jù)shuffle的數(shù)據(jù)量來(lái)設(shè)置partition數(shù)，偏大些即可 大量的rdd在union時(shí)使用new unionrdd(sc, seq(rdd)，防止 stackoverflowerror，而不是a.union(b).union(c) 使用yarn時(shí)設(shè)置spark.shuffle.service.enabled=true，減少shuffle數(shù) 據(jù)重做的代價(jià) 將akka的參數(shù)時(shí)間設(shè)長(zhǎng)一些以及配置gc參數(shù) 對(duì)于etl作業(yè)使用勱態(tài)資源擴(kuò)縮容特性 問(wèn)題問(wèn)題: 在此這前,spark仸務(wù)在task執(zhí)行前通過(guò)指定參數(shù)來(lái)啟勱固定數(shù)目的executor, 一旦達(dá)到用戶指定值后保持住這個(gè)資源數(shù),即使仸務(wù)運(yùn)行過(guò)程中的task數(shù)增加戒 減少,也不會(huì)去改變executor的資源數(shù)目 executor task executor task sparkcontext task task job executorallocationmanager spark.dynamicallocation.enabled true spark.dynamicallocation.executoridletimeout 120 spark.dynamicallocation.schedulerbacklogtimeout 10 spark.dynamicallocation.minexecutors/maxexecutors redcue階段在map階段未全部完成前執(zhí)行 executor-1 maptask-1 maptask-2 maptask-3 maptask-4 reducetask-1 reducetask-2 reducetask-3 reducetask-4 reducetask-5 time 1 time 2 time 3 問(wèn)題問(wèn)題:當(dāng)申請(qǐng)固定的executors時(shí)且task數(shù)大于executor數(shù)，這時(shí)存在資源的空閑狀態(tài) executor-2 executor-3 executor-1 maptask-1 maptask-2 maptask-3 maptask-4 free free reducetask-1 reducetask-2 reducetask-3 executor-2 executor-3 reducetask-4 reducetask-5 time 4 當(dāng)當(dāng)executor沒(méi)有沒(méi)有map task需要執(zhí)行時(shí)，即可在上面運(yùn)行需要執(zhí)行時(shí)，即可在上面運(yùn)行reduce task并提前去拉并提前去拉 已完成的已完成的map數(shù)據(jù)。數(shù)據(jù)。 stage-1(task.size=parallelism) 基于數(shù)據(jù)的大小預(yù)測(cè)stage的partition數(shù) 怎么去設(shè)置spark.default.parallelism/spark.sql.shuffle.partitions? 輸入數(shù)據(jù)每天在變化且有大量的歷叱hivesql都沒(méi)有設(shè)置partition數(shù) maptask maptask maptask stage-0(task.size=blocks.size) reducetask reducetask reducetask reducetask reducetask reducetask maptask maptask maptask stage-0(task.size=blocks.size) reducetask reducetask reducetask reducetask reducetask reducetask stage-1(task.size=stage-0.input/256m) stage-2(task.size=stage-1.input/256m) stage-2(task.size=parallelism) 問(wèn)題問(wèn)題: 為sparksql的每個(gè)session分配一個(gè)driver yarn/applicationmaster querymaster spark driver sparksqlclient shell jdbc/odbc executor executor spark driver shell jdbc/odbc executor executor query-1 query-2 hive server query-1 query-2 hive server sparksqlclient sparksql hivecontext spark driver 問(wèn)題問(wèn)題: 一個(gè)hiveserver上所有的query都使用一個(gè)driver，導(dǎo)致query的內(nèi)存和資源都會(huì)受 到單個(gè)driver的限制 sparksql hivecontext sqlpa