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

騰訊在spark上的應用與實踐優(yōu)化 王聯(lián)輝 agenda spark在騰訊的當前現(xiàn)狀 spark在騰訊的典型應用及效果 騰訊在spark上的實踐和優(yōu)化 未來工作和計劃 tdw(tencent distributed data warehouse) overview mapreduce spark hive/pig ide lhoste(workflow system) gaia(based on yarn) gaia集群結(jié)點數(shù): 8000+ hdfs的存儲空間: 150pb+ 每天新增數(shù)據(jù): 1pb+ 每天仸務數(shù): 1m+ 每天計算量: 10pb+ user defined jobs (scala/java/python/) hdfs storm docker ide: 用于提交sql戒腳本的eclipse插件和web界面 lhoste: 各類作業(yè)的工作流調(diào)度系統(tǒng)，類似于oozie gaia: 基于yarn進行定制和優(yōu)化的資源管理系統(tǒng) user 為什么我們引入spark? dag計算引擎 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在騰訊的當前現(xiàn)狀 作業(yè)類型：etl,sparksql,machine learning,graph compute,streaming 每天仸務數(shù)：10k+ 部署模式：gaia(8000+ nodes,with 24 cores and 60g memory each) 底層存儲: 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在騰訊的典型應用及效果 case 1: 預測用戶的廣告點擊概率 case 2: 計算二個好友間的共同好友數(shù) case 3: 用于etl的sparksql和dag任務 case 1: 預測用戶的廣告點擊概率 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ù)據(jù)并推送到hdfs上 將訓練數(shù)據(jù)(200g+) 導入至rdd & cache 初始化隨機值給w并廣播至各個executor 迭代訓練60次左史 for (i p.x * (1 / (1 + exp(-p.y * (w.dot(p.x) - 1) * p.y .reduce(_ + _) w -= gradient 將模型推送廣告后臺在線服務器 每個計算的時間在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: 預測用戶的廣告點擊概率 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: 計算二個好友間的共同好友數(shù) table-1 table-2 case 2: 計算二個好友間的共同好友數(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: 計算二個好友間的共同好友數(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ù)量來確定partition數(shù) 使用sort-based shuffle來提升性能和減少reduce的內(nèi)存使用(spark-1.2). 在大集群時當連接超時后選擇重試來減少executor丟失的概率(spark-1.2). 為避免executor被yarn給kill,通常需要設(shè)置 spark.yarn.executor.memoryoverhead. case 2: 計算二個好友間的共同好友數(shù) case 3: 用于etl的sparksql和dag仸務 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上的實踐和優(yōu)化 應用程序開發(fā)中的使用經(jīng)驗 對于etl作業(yè)使用勱態(tài)資源擴縮容特性 redcue階段在map階段未全部完成前執(zhí)行 基于數(shù)據(jù)的大小預測stage的partition數(shù) 為sparksql的每個session分配一個driver count(distinct)的優(yōu)化 基于排序的groupby/join 應用程序開發(fā)中的使用經(jīng)驗 當有小表時使用broadcast join代替common join 盡量使用reducebykey來代替groupbykey 設(shè)置spark.serializer=org.apache.spark.serializer.kryoserializer 根據(jù)shuffle的數(shù)據(jù)量來設(shè)置partition數(shù)，偏大些即可 大量的rdd在union時使用new unionrdd(sc, seq(rdd)，防止 stackoverflowerror，而不是a.union(b).union(c) 使用yarn時設(shè)置spark.shuffle.service.enabled=true，減少shuffle數(shù) 據(jù)重做的代價 將akka的參數(shù)時間設(shè)長一些以及配置gc參數(shù) 對于etl作業(yè)使用勱態(tài)資源擴縮容特性 問題問題: 在此這前,spark仸務在task執(zhí)行前通過指定參數(shù)來啟勱固定數(shù)目的executor, 一旦達到用戶指定值后保持住這個資源數(shù),即使仸務運行過程中的task數(shù)增加戒 減少,也不會去改變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 問題問題:當申請固定的executors時且task數(shù)大于executor數(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 當當executor沒有沒有map task需要執(zhí)行時，即可在上面運行需要執(zhí)行時，即可在上面運行reduce task并提前去拉并提前去拉 已完成的已完成的map數(shù)據(jù)。數(shù)據(jù)。 stage-1(task.size=parallelism) 基于數(shù)據(jù)的大小預測stage的partition數(shù) 怎么去設(shè)置spark.default.parallelism/spark.sql.shuffle.partitions? 輸入數(shù)據(jù)每天在變化且有大量的歷叱hivesql都沒有設(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) 問題問題: 為sparksql的每個session分配一個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 問題問題: 一個hiveserver上所有的query都使用一個driver，導致query的內(nèi)存和資源都會受 到單個driver的限制 sparksql hivecontext sqlpa