腾讯在Spark上的应用与实践优化

腾讯在SPARK上的应用与实践优化王联辉AGENDASPARK在腾讯的当前现状SPARK在腾讯的典型应用及效果腾讯在SPARK上的实践和优化未来工作和计划TDWTENCENTDISTRIBUTEDDATAWAREHOUSEOVERVIEWMAPREDUCESPARKHIVE/PIGIDELHOSTEWORKFLOWSYSTEMGAIABASEDONYARNGAIA集群结点数8000HDFS的存储空间150PB每天新增数据1PB每天仸务数1M每天计算量10PBUSERDEFINEDJOBSSCALA/JAVA/PYTHON/HDFSSTORMDOCKERIDE用于提交SQL戒脚本的ECLIPSE插件和WEB界面LHOSTE各类作业的工作流调度系统，类似于OOZIEGAIA基于YARN进行定制和优化的资源管理系统USER为什么我们引入SPARKDAG计算引擎DATACACHESHARING其他TASK调度，数据广播等MMMRRHDFSMMRHDFSMMRHDFSMMRMMMRRMMRRRHDFSITER1ITER2ITER3RESULT1RESULT2RESULT3HDFSITER1ITER2ITER3RESULT1RESULT2RESULT3SPARK在腾讯的当前现状作业类型ETL,SPARKSQL,MACHINELEARNING,GRAPHCOMPUTE,STREAMING每天仸务数10K部署模式GAIA8000NODES,WITH24CORESAND60GMEMORYEACH底层存储HDFS/HIVE/HBASE/MYSQL/从2013年的SPARK06版本开始，目前的版本是SPARK12SPARKSPARKSQLSPARKSTREAMINGMLLIBMACHINELEARNINGGRAPHXSCALA/JAVA/PYTHONETLHDFSHIVETABLEHBASEMYSQLPOSTGRESQLOTHERSSPARK在腾讯的典型应用及效果CASE1预测用户的广告点击概率CASE2计算二个好友间的共同好友数CASE3用于ETL的SPARKSQL和DAG任务CASE1预测用户的广告点击概率GENDERAGEMARITALSTATUSLOCATIONADONEADTWOISCLICKMAN20UNMARRIEDNEWYORKTRUEFALSEYESWOMAN40MARRIEDCALIFORNIAFALSETRUEYESMAN60MARRIEDCALIFORNIAFALSETRUENOGENDERAGEMARITALSTATUSLOCATIONADONEADTWOCLICKSPOSSIBILITYMAN30UNMARRIEDNEWYORKTRUEFALSEMAN30UNMARRIEDNEWYORKFALSETRUEGENDERAGEMARITALSTATUSLOCATIONADONEADTWOCLICKSPOSSIBILITYMAN30UNMARRIEDNEWYORKTRUEFALSE30MAN30UNMARRIEDNEWYORKFALSETRUE50WMANWOMANAD1AD20896089504900489MODELTRAINATSPARKTABLE1TABLE2TABLE3TABLE4实时预处理训练数据并推送到HDFS上将训练数据200G导入至RDDCACHE初始化随机值给W并广播至各个EXECUTOR迭代训练60次左史FORIPX1/1EXPPYWDOTPX1PYREDUCE_WGRADIENT将模型推送广告后台在线服务器每个计算的时间在1015分钟HDFSDRIVEREXECUTORCACHETASK/RDDEXECUTORCACHETASK/RDDTASK/RDDTASK/RDDTASK/RDDWWWWGRADIENTGRADIENTGRADIENTDATACOLLECTTOOLSCASE1预测用户的广告点击概率USERFRIEND2141125363655636USERFRIENDNUMBEROFMUTUALFRIENDS2104101205316306515613605亿用户数，500亿条边10亿用户数，1000亿条边USERLINKSTABLEID,FIDUSERFRIENDSTABLEID,FIDSGROUPBYTEMP1TABLEID,FID,IDFIDSJOINBYTEMPTABLEID,FID,IDFIDS,FIDFIDSJOINBYRESULTTABLEID,FID,COUNTCASE2计算二个好友间的共同好友数TABLE1TABLE2CASE2计算二个好友间的共同好友数USERFRIEND2141125363655636PIDIDFID12112236341536346556VALPARTITIONSTRATEGYPARTITIONSTRATEGYEDGEPARTITION2DTABLE1TABLE2PIDIDFID12112236341536346556IDPIDLIST11,32132,34353,462,3,4PIDUIDFRIENDLIST1122123663,5312364153,663,5453,663,5IDFRIENDLIST1221364153,663,5IDPIDLISTFRIENDLIST11,3221132,3643153,43,662,3,43,5JOINBYIDJOINBYPIDUSERFRIENDTOTALNUMBEROFMUTUALFRIENDS210410CASE2计算二个好友间的共同好友数TABLE1TABLE2TABLE3TABLE4TABLE5TABLE6EDGES/VERTEXNODESHIVEMAPREDUCESPARKEXECUTORSMEMORY,CORES/PARTITIONS50B/05B20012H2H15G/1/500050B/05B100043M15G/1/5000100B/1B20024H6H25G/1/10000100B/1B10002H25G/1/10000ONENODE,55GMEMORY,212TSATADISK,2419GHZCPU,1GBPSNETWORK根据SHUFFLE的数据量来确定PARTITION数使用SORTBASEDSHUFFLE来提升性能和减少REDUCE的内存使用SPARK12在大集群时当连接超时后选择重试来减少EXECUTOR丢失的概率SPARK12为避免EXECUTOR被YARN给KILL,通常需要设置SPARKYARNEXECUTORMEMORYOVERHEADCASE2计算二个好友间的共同好友数CASE3用于ETL的SPARKSQL和DAG仸务INSERTTABLETEST_RESULTSELECTT3D,COUNTFROMSELECTDISTINCTA,BFROMJOIN_1T1JOINSELECTDISTINCTB,CFROMJOIN_2T2ONT1AT2CJOINSELECTDISTINCTC,DFROMJOIN_3T3ONT2BT3DGROUPBYT3DCOMPUTERESOURCESTIMEHIVE200MAPS,50REDUCES1CORE,32GMEMORY30MINSPARKSQL50EXECUTORS1CORE,4GMEMORY5MIN腾讯在SPARK上的实践和优化应用程序开发中的使用经验对于ETL作业使用劢态资源扩缩容特性REDCUE阶段在MAP阶段未全部完成前执行基于数据的大小预测STAGE的PARTITION数为SPARKSQL的每个SESSION分配一个DRIVERCOUNTDISTINCT的优化基于排序的GROUPBY/JOIN应用程序开发中的使用经验当有小表时使用BROADCASTJOIN代替COMMONJOIN尽量使用REDUCEBYKEY来代替GROUPBYKEY设置SPARKSERIALIZERORGAPACHESPARKSERIALIZERKRYOSERIALIZER根据SHUFFLE的数据量来设置PARTITION数，偏大些即可大量的RDD在UNION时使用NEWUNIONRDDSC,SEQRDD，防止STACKOVERFLOWERROR，而不是AUNIONBUNIONC使用YARN时设置SPARKSHUFFLESERVICEENABLEDTRUE，减少SHUFFLE数据重做的代价将AKKA的参数时间设长一些以及配置GC参数对于ETL作业使用劢态资源扩缩容特性问题在此这前,SPARK仸务在TASK执行前通过指定参数来启劢固定数目的EXECUTOR,一旦达到用户指定值后保持住这个资源数,即使仸务运行过程中的TASK数增加戒减少,也不会去改变EXECUTOR的资源数目EXECUTORTASKEXECUTORTASKSPARKCONTEXTTASKTASKJOBEXECUTORALLOCATIONMANAGERSPARKDYNAMICALLOCATIONENABLEDTRUESPARKDYNAMICALLOCATIONEXECUTORIDLETIMEOUT120SPARKDYNAMICALLOCATIONSCHEDULERBACKLOGTIMEOUT10SPARKDYNAMICALLOCATIONMINEXECUTORS/MAXEXECUTORSREDCUE阶段在MAP阶段未全部完成前执行EXECUTOR1MAPTASK1MAPTASK2MAPTASK3MAPTASK4REDUCETASK1REDUCETASK2REDUCETASK3REDUCETASK4REDUCETASK5TIME1TIME2TIME3问题当申请固定的EXECUTORS时且TASK数大于EXECUTOR数，这时存在资源的空闲状态EXECUTOR2EXECUTOR3EXECUTOR1MAPTASK1MAPTASK2MAPTASK3MAPTASK4FREEFREEREDUCETASK1REDUCETASK2REDUCETASK3EXECUTOR2EXECUTOR3REDUCETASK4REDUCETASK5TIME4当EXECUTOR没有MAPTASK需要执行时，即可在上面运行REDUCETASK并提前去拉已完成的MAP数据。STAGE1TASKSIZEPARALLELISM基于数据的大小预测STAGE的PARTITION数怎么去设置SPARKDEFAULTPARALLELISM/SPARKSQLSHUFFLEPARTITIONS输入数据每天在变化且有大量的历叱HIVESQL都没有设置PARTITION数MAPTASKMAPTASKMAPTASKSTAGE0TASKSIZEBLOCKSSIZEREDUCETASKREDUCETASKREDUCETASKREDUCETASKREDUCETASKREDUCETASKMAPTASKMAPTASKMAPTASKSTAGE0TASKSIZEBLOCKSSIZEREDUCETASKREDUCETASKREDUCETASKREDUCETASKREDUCETASKREDUCETASKSTAGE1TASKSIZESTAGE0INPUT/256MSTAGE2TASKSIZESTAGE1INPUT/256MSTAGE2TASKSIZEPARALLELISM问题为SPARKSQL的每个SESSION分配一个DRIVERYARN/APPLICATIONMASTERQUERYMASTERSPARKDRIVERSPARKSQLCLIENTSHELLJDBC/ODBCEXECUTOREXECUTORSPARKDRIVERSHELLJDBC/ODBCEXECUTOREXECUTORQUERY1QUERY2HIVESERVERQUERY1QUERY2HIVESERVERSPARKSQLCLIENTSPARKSQLHIVECONTEXTSPARKDRIVER问题一个HIVESERVER上所有的QUERY都使用一个DRIVER，导致QUERY的内存和资源都会受到单个DRIVER的限制SPARKSQLHIVECONTEXTSQLPARSER,LOGICALPLAN,OPTIMIZER,PHYSICALPLANCOUNTDISTINCT的优化语句SELECTA,COUNTDISTINCTBFROMTABLE1GROUPBYA当B列的数据出现倾斜时，语句出现OOMSTAGE11,21,21,42,51,21,42,51,31,31,42,61,31,42,61,41,41,32,61,22,11,22,51,12,11,32,2STAGE2STAGE3HASHSHUFFLEA,BHASHSHUFFLEALOCALDISTINCTLOCALDISTINCTLOCALCOMBINER将该语句的执行计划改为如图所示LOCALCOMBINER基于排序的G