四两拨千斤:借助Spark GraphX将QQ千亿关系链计算提速20倍 | 数盟社区

原作者: Andy  来自: 腾讯大数据

 

腾讯QQ有着国内最大的关系链,而共同好友数,属于社交网络分析的基本指标之一,是其它复杂指标的基础。借助Spark GraphX,我们用寥寥100行核心代码,在高配置的TDW-Spark集群上,只花了2个半小时,便完成了原来需要2天的全量共同好友计算。这标志着QQ千亿级别的关系链计算进入了小时级别时代,并具备复杂图模型的快速计算能力。
问题描述
共同好友数可以用于刻画用户与用户间的关系紧密程度,包括 陌生人/熟人分析,好友亲密度,好友推荐,社团划分等各个方面,是社交网络分析的最基础指标。其计算逻辑非常简单明了。为了简化模型和降低计算量,这里加了几个约束:
只有好友之间才进行计算
好友关系是有向的
不关注具体的好友

显而易见,用户5和6的共同好友数为4。这个计算貌似非常简单,但是当图的规模扩展到腾讯的级别:用户数(点)为十亿级别,关系数(边)为千亿级别时,那这个问题就一点都不简单了。
大致估算一下,假设每个节点平均的好友数是100,每个点id为Long型,占用8个字节,如果用普通Join计算的话,那么中间的数据量大概是1 billion* 800*800B=640TB,需要通过网络传输640TB的数据,这个量非常的恐怖了,想在一个SQL中完成,几乎是不可能的。原来旧的计算方式,只能通过分而治之的方法实现。通过把关系链表,按号段拆分60份,分别连接用户好友全量表,分成多个SQL任务运行。这样做每个SQL任务都需要载入一次全量关系链,磁盘 I/O 时间严重拖慢计算进度,整个过程需要耗费超过两天的计算时间。
其实共同好友数是典型的图问题之一,因此很自然的,我们想到了引入新的图计算框架和模型,来优化计算过程,让这个过程更加高效和科学。
框架选择
目前的分布式图计算框架,并没有太多好的选择,在过去一年半中,业界的分布式图计算框架,进展基本停滞。经过反复选择,我们还是选择了GraphX,主要原因有如下3个:
进展
虽然GraphX本身没什么进展,但是Spark本身的发展很快,从1.4到1.6版本,Spark Core在性能和稳定性上有了不少的提升。GraphX某种程度上,多多少少还是得到了好处
语义
GraphX的语义和运算符相对丰富,可以进行比较好的图算法描述,适合变化多样的图需求
门槛
GraphX的最大消耗是内存,某种程度上,这是个比较低门槛的投入,可以在预期内得到解决
基于这样的考虑,我们选择了GraphX作为共同好友算法的底层框架,并从软件和硬件两方面入手,进行了一系列的优化。
模型简化
基于GraphX的图模型思想,我们进行数好友模型的简化。整个过程分为两个阶段:
Phrase1——找邻居
这个阶段,其实就是一放一收,和Map-Reduce模型有异曲同工之妙,分3步
1. 每个顶点,将自己的id,发送给自己所有的邻居
2. 每个顶点,将收到的所有邻居id,合并为一个List
3. 对新List进行排序,并和原来的图进行关联,附到顶点之上

这个阶段,使用GraphX的aggregateMessages,定义好sendMsg和mergeMsg的方法,就可轻松实现。
Phrase2——数好友
这个阶段,只要一步,但是非常的关键,而且计算量也非常大,需要充分的利用了Triplet的特性
1. 遍历所有的Triplet,对2个好友的有序好友List进行扫描匹配,数出共同好友数,并将其更新到edge之上

这个阶段,需要充分利用GraphX的Triplet特性。需要为了实现Triplet功能,GraphX在设计上消耗了很多的内存,无论我们是否使用,它都在那里,静静的占用着内存,所以我们要充分利用好它,将数好友的过程,简化为一次Triplet的遍历。
如果没有这个特性,为了数好友之间的共同好友,就要把一个好友的所有一度好友,发送到它所有的邻居之上,这样的消息广播量,是非常巨大的,会形成消息风暴,相当于计算二跳邻居。根据我们在腾讯数据量上的测试,这个在现有的硬件下,是无法实现的。
这2个阶段经过最终优化之后,代码非常的简洁,只要聊聊的数十行代码,便完成了20亿级别的点,千亿级别的边的共同好友计算,可谓于无声处听惊雷。将众多的技术难点,都通过GraphX的优化,化解消弭于无形之中。由于产品的需求,这个计算需要是精确数,而不能是近似值,在数好友的过程中,很多的优化方法,不能被用上,否则的话,可以进一步的提升速度。
硬件选择
在分布式系统中,往往会倾向于用大量的小低配机器,来完成巨大的计算任务。其思想是即便再复杂的计算,只要将大数据,分解为足够小的数据片,总能在足够多的机器上,通过性能的降低和时间的拖延,来完成计算任务。但是很遗憾,在图计算这样的场景下,尤其是GraphX的设计框架,这个是行不通的。
要发挥GraphX的最佳性能,最少要有128G以上的内存
主要原因有两个是:
节点复制——越小越浪费
GraphX使用了点切割的方式,这是一种用空间换时间的方法,通过将浪费一定的内存,将点和它的邻居放到一起,减少Executor之间的通信。
如果用小内存的Executor来运行图算法,假设1个节点,需要10个Executor才能放下它的邻居,那么它就需要被复制10份,才能进行计算。如果用大内存的Executor,1个Executor就能放下它的所有邻居,理论上它就只需要被复制一次,大大减少空间占用。
节点膨胀——越小越慢
图计算中,常常会进行消息的扩散和收集,并把最终的结果,汇总到单个节点之上。
以共同好友数模型为例,第一步需要将节点的一跳好友都收集到该节点上。即便根据邓巴的“150定律”,将一跳好友的个数,限制在150之内。那么图的占用空间,还是很可能会膨胀150倍。
这个时候,如果内存空间不够的话,GraphX为了容下所有的数据,会需要在节点之间,进行大量的Shuffle和Spill操作,使得后续的计算,变得非常慢。
其实这两个问题,在Spark的其它机器学习算法中,或多或少都会有,也是分布式计算系统中,经常面临的问题。但是在图计算中,它们是无法被忽略的问题,而且非常的严重。所以,这决定了GraphX需要大的内存,才能有良好的性能。
在正常情况下,128G内存,减掉8G的系统占用,剩下120G。这时配置每个Executor 60G内存,2个Core,每个Core分到30G的内存。这时不需要申请太多的Executor,经过合理的性能优化,全量关系链计算,可以运行成功。
性能优化
即便有了良好的模型和硬件保障,在面对QQ如此巨型的关系链时,依然需要熟练运用GraphX的技巧,并避开各种雷区,才能最终到达终点。简单总结两点:
图缓存:To Cache or not to Cache? That is a question
Spark和GraphX原本设计的精妙之处,亮点之一,便在于Cache,也就Persist(MEMORY_ONLY),或者Persist(MEMORY_AND_DISK)。可以把RDD和Graph,Cache到内存中,方便多次调用而无需重新计算。那么是否对所有的RDD,或者图,都Cache一下,是最佳的选择呢?答案是未必。
判断是否要Cache一个Graph或RDD,最简单和重要的标准,就是
该Graph,是否会在后续的过程中,被直接使用多次,包括迭代。
如果会,那么这个Graph就要被Persist,然后通过action触发
如果不会,那么反过来,最好把这个Graph直接unpersist掉
一个Graph被Cache的话,一般最终体现为2个RDD的Cache,一个是Edge,一个是Vertex,其占用量是非常巨大的。在整体空间有限的情况下,cache会导致内存的使用量大大加剧,引发多次GC和重算,反而会拖慢速度。在QQ全量的关系链计算,一个全量图是非常大的,因此如果在一个图没被多次使用,那么先unpersist,再返回给下一个计算步骤,反而成了最佳实践。
示例代码如下:
val oneNbrGraph = computeOneNbr(originalGraph)
oneNbrGraph.unpersist()
val resultRdd = oneNbrGraph.triplets.map {
    …………
}
当然,既然unpersist了,切记 它只能被再用一次了。
分区策略:EdgePartition2D
对GraphX有所了解的人,应该都知道,有4种分区的策略,而其中性能最好的,莫过于EdgePartition2D这种边分区策略。但是由于QQ全量的关系链非常的大,所以,如果先用默认策略,构造了图,再调用partitionBy的方法来改变分区策略,那么会多一步代价非常高的计算。
因此,为了减少不必要的计算步骤,我们建议在构造图之前,先对Edge使用该策略进行划分,再用划分好的Edge RDD,进行图构建。 示例代码如下:
val edgesRepartitionRdd = edgeRdd.map { case (src, dst) => val pid = PartitionStrategy.EdgePartition2D.getPartition(src, dst, partitionNum) (pid, (src, dst)) }.partitionBy(new HashPartitioner(partitionNum)).map { case (pid, (src, dst)) => Edge(src, dst, 1) }
val g = Graph.fromEdges(
  edgesRepartitionRdd,
  ...
)
当然,有个非常重要的Hint别忘记:partitionNum必须是平方数,才能达到最佳的性能。
最终效果
经过反复多次优化之后,在高配置集群的资源充足情况下,使用如下配置项:
|配置项|配置值| |—|—| |executors | 360 | |executor-cores | 2 | |executor-memory | 50g | |并行度 | 10000 |
总共消耗18T的内存,可以在2个半小时左右,完成整个QQ关系链的计算。BTW:
这个配置其实只是可运行配置,并非最佳配置。如果内存有512G,每个Executor配到80G,每个机器6个Executor,每个Executor 4个Core,应该能有更好的表现。可以进一步减少Executor的数目。
集群硬盘是SATA盘,而非SSD硬盘。在整个计算过程中,由于QQ的关系链实在庞大,中间的过程也比较复杂,所以不推荐用MEMORYONLY的选项,一旦内存充不下,重算的代价非常高,所以建议用MEMORYAND_DISK。而但是用这种方式意味着和硬盘的通信是必不可少的,因此如果硬盘能换成SSD的高性能盘,对整体性能会有很大提升。
在集群处于正常负荷的情况下,资源充足时,GraphX的任务不发生重跑时,作业可以在2小时10分之内,完成全量计算。但这是在运气最佳,没有任何Task发生重跑的情况下的表现。一旦有任务Task失败,Spark会自动重跑,但是整个计算过程会变得非常长,即便是很少的2-3个Task失败,也会将计算过程,延长到3个多小时甚至更多,这是因为GraphX的Failover没做好,而且在有多次迭代的时候,这个现象会更加严重。
总结和展望
整个的优化过程,貌似风轻云淡,但是中间经过了反复调优,多次在0.1的抽样数据和1.0的全量数据之间切换,优化每一步的操作,将硬件和GraphX的性能压榨到极致,才最终得到这个结果。
在这个过程中,我们发现无论应用层再怎么优化,核心层的软肋,始终制约着上层算法的性能。包括 Graph中最大限度的预创建和 RDD Cache的激进使用等问题,都会导致性能和稳定性不足,使得很多算法在腾讯级别的图数据下,显得捉襟见肘。其实这也难怪,GraphX的代码,从1.3版本开始,便已经一直没有变动,基本是在吃Core优化的红利,沾光提高性能,没有任何实质性的改进,如果要继续使用,在核心上必须有所提升才行。
腾讯作为拥有国内最大的关系链,在图计算的领域,无论是处于框架,模型,还是存储,都大有可为,可以做很多的事情。腾讯的数据平台部,作为公司的大数据支撑平台,欢迎在这方面有兴趣的业界同仁,和我们进行更多的合
注:转载文章均来自于公开网络,仅供学习使用,不会用于任何商业用途,如果侵犯到原作者的权益,请您与我们联系删除或者授权事宜,联系邮箱:contact@dataunion.org。转载数盟网站文章请注明原文章作者,否则产生的任何版权纠纷与数盟无关。
期待你一针见血的评论,Come on!

不用想啦,马上 "登录"  发表自已的想法.