腾讯TDW千台Spark千亿节点对相似度计算

相似度计算在信息检索、数据挖掘等领域有着广泛的应用，是目前推荐引擎中的重要组成部分。随着互联网用户数目和内容的爆炸性增长，对大规模数据进行相似度计算的需求变得日益强烈。在传统的MapReduce框架下进行相似度计算会引入大量的网络开销，导致性能低下。我们借助于Spark对内存计算的支持以及图划分的思想，大大降低了网络数据传输量；并通过在系统层次对Spark的改进优化，使其可以稳定地扩展至上千台规模。本文将介绍腾讯TDW使用千台规模的Spark集群来对千亿量级的节点对进行相似度计算这个案例，通过实验对比，我们优化后的性能是MapReduce的6倍以上，是GraphX的2倍以上。

一、介绍

相似度是指两个节点之间特定属性的相似程度，相似度计算是数据挖掘、推荐引擎中的最基本问题。例如在推荐系统中通过计算推荐物品的相似度，从而给目标用户推荐与他喜欢的物品相似度较高的物品，或是计算用户之间的相似度，给目标用户推荐与其相似的用户喜欢的物品。因此，相似度计算技术在很大程度上决定着推荐系统的性能。

随着大数据时代的来临，日益增加的数据量使得单机的计算能力已经远远无法满足需求。在对大规模的节点对进行相似度计算时，分布式处理往往是可行的解决方案。MapReduce是目前流行的分布式编程框架。Hadoop与Spark是MapReduce编程模型的两个开源实现。相比于Hadoop，Spark提供了cache机制，增加了对迭代计算的支持；还提供了DAG调度来支持复杂的计算任务，减少了中间结果的磁盘读写，能够获得更佳的性能。

本文将介绍腾讯TDW使用Spark来对千亿量级的节点对进行相似度计算的案例研究，我们在计算方法和系统两个层次都进行了改进优化，获得性能提升的同时，还具备了千台集群的扩展能力。在本文接下来的内容中，先简要介绍相似度计算这个问题，接着介绍在Hadoop和Spark上的两种实现方式，并通过在两个数据集上的实验进行对比分析，最后进行总结。

二、问题描述

输入数据可以表示成两张表：

1.节点关系表relation，字段有id, fid，表示两个节点存在关系。

2.节点特征表features，字段有id, feature，表示每个节点具有的特征信息。

下列两个表格表示了在一个拥有6个节点的关系网络中，节点关系表和节点特征表的情况。

相似度计算即是对节点关系表中的所有节点对 (id，fid)，其特征向量分别为 和,利用相似度计算函数similarity-Calculation，计算和之间的相似度。相似度计算函数similarity-Calculation依据具体的相似度衡量方法而定。

三、MapReduce 解决方案

Hive是建立在Hadoop之上提供SQL接口处理的海量数据处理工具，对于上述相似度计算问题，其计算流程可以用如下SQL来描述，并使用Hive来计算。

整个计算流程可以分为两个步骤：

1.通过两次JOIN操作，生成一张临时表，临时表中的一个元组对应节点关系表中的一对节点和这两个节点的特征向量。

2. 遍历临时表，对每个元组中的两个节点计算其相似度。

下图展示了该SQL语句的执行过程：

使用Hive对千亿节点关系记录进行相似度计算，两次JOIN操作成为性能的主要瓶颈瓶颈。在两次JOIN的过程中，网络数据传输和磁盘读写达到了200TB，集群多数结点的硬盘无法支持，任务失败经常发生，作业运行了时间超过了24小时。通过将节点关系表拆分成多个子表，每个子表独立地进行相似度计算，多个子表的任务并行执行，最后再将多个子作业的结果汇总，得到最终结果。采用这样的方式，作业总时间仍然超过了24小时。

四、Spark解决方案

通过对Hive计算过程的分析，我们发现网络数据开销主要来自于节点特征向量的大量复制。对于节点关系表中的每对关系，计算时都需要得到两个节点的特征向量，从而导致了大量的数据复制。因此，我们从两个方面去减少数据复制：

1.采用二维图划分的思想，减少节点的复制数目

2.每个数据分区中，对于同一个节点，只保留一份该节点特征向量

二维图划分方法

任何一张关系网络，都可以用一个大矩阵M来表示，矩阵的两个维度用来表示节点，矩阵的元素M[i, j]表示节点i和节点j是否存在关联，如果存在，则M[i,j]值为1，否则，M[i, j]值为0。下图展示了通过采用二维划分的方法，将一个矩阵划分成了16个分区。

使用二维划分可以减少节点的复制数目。假设分区总数为，采用一维划分的方法，最差情况下每个节点的复制份数是，即每个分区都会有该节点的复制；采用二维划分方法，最差情况下每个节点的复制份是 。对于大数据量，分区总数通常很大，所以采用二维划分通常可以减少每个节点的复制份数。

计算步骤

1.利用二维划分方法将节点关系表划分成多个数据分区，假设我们将分区数设为4，则Table 1所示的节点关系表将会划分到4个分区，每个元组对应的分区如下Table 3所示：

2.根据每个分区中的节点列表，计算出每个节点所在的分区列表，称为路由表，记录了每个节点所在的分区信息，其结果如Table 4所示。

3.根据路由表将每个节点的特征向量发送至每个分区之中，保证每个分区中一个节点只保存一份特征向量，如Table 5所示。

4.对于每个分区，将该分区的关系集合与该分区中所有结点的特征向量进行关联，遍历每对节点关系，利用相似度函数和特征向量计算二者的相似度。

通过以上步骤，即可以计算出节点关系表中每对节点的相似度。与MapReduce的计算方法相比，如果一个用户多次出现在同一个分区中，比如用户1在分区1中出现了两次，上述计算步骤只会将用户1的特征向量发送一份到分区1中，但是MapReduce的计算方法会发送两次，产生冗余的网络数据传输。使用上述计算方法，我们将网络传输量降到了50 T，远小于MapReduce方法的网络传输量。

系统层次优化

除了在计算流程上进行改进，我们还对Spark进行了以下方面的优化：

1）优化分区参数设置。在相似度计算的应用中，分区个数越多，会导致节点的复制份数增加，从而增大网络数据传输量。因此我们基于中间结果的统计信息来确定确定分区个数，使得在充分利用每个节点内存和CPU的前提下，最小化分区个数。

2）优化内存表示。由于数据量大，对象个数多，导致内存使用量较高，GC时间较长。我们使用列存储格式来对内存数据进行压缩，减少数据量的同时也减少了对象个数。

3）提高网络稳定性。随着集群中机器数目的增加，网络连接数也会成倍增加。当网络出现拥挤时，经常会伴随着连接超时从而导致shuffle数据拉取失败。更糟糕的情况是，网络超时会让Master误认为Executor已经丢失，故会使得整个Executor上已经完成的任务全部重做。因此在shuffle时增加网络超时重试机制，同时控制每次发送的请求连接数，避免shuffle拉数据超时，减少任务失败次数，防止Executor丢失的情况出现。

4）使用sort-based shuffle时将文件块索引信息缓存一份在内存中，后续拉数据时直接读内存获取索引信息。预测执行时，当同一任务的一批运行实例有一个完成时，杀掉正在运行的其余实例，提早释放计算资源。

5）参数调整。由于每个Executor进程还会使用到堆外内存，因此Executor进程占用的内存往往会大于JVM设定的最大值，为了保证Gaia不会将超过JVM内存的Executor进程杀掉，配置参数spark.yarn.executor.memoryOverhead以免被kill。由于Executor在Full GC时需要较长时间，需要配置参数spark.storage.blockManagerSlaveTimeoutMs来延长blockManager的超时时间。

五、实验对比

实验环境：

我们分别在拥有200台、600台和1000台TS5机器节点的集群上进行了对比，每台机器拥有64GB内存，2*12T硬盘，24线程CPU。

我们在两个数据集上进行了Hadoop、社区GraphX和TDW-Spark的性能对比，一个数据集拥有五百亿节点对，而另一个拥有千亿量级的节点对。实验结果如下表所示:

通过上述实验对比，可以看出在MapReduce上的实现的性能远远低于在Spark上的性能，使用JOIN的方法使得网络通信开销非常大，五百亿数据集的任务执行时间超过12个小时，千亿数据集任务执行时间超过24个小时；GraphX采用的同样是二维图划分，但是由于其是一个面向通用的图计算框架，维护了复杂的数据结构和计算流程，造成性能下降。同时，GraphX在网络稳定性方面存在许多问题，当集群规模达到600台时便会有大量的任务失败。

与前两者相比，TDW-Spark在集群为200台时在两个数据集上都获得了较大的性能增长，所消耗时间少于GraphX的一半。当集群规模从200台扩充至600台，TDW-Spark在五百亿节点对数据集上获得加速比218%，在千亿节点上的加速比为280%；当集群规模从200台扩充至1000台时，加速比分别为279%和350%。因此，TDW-Spark不仅在性能上获得了很大的提升，还可以在千台规模的集群之上稳定运行，同时获得良好的水平扩展能力。

六、结论

Spark是目前Apache中最活跃的开源项目之一，已经形成了一套成熟的大数据处理生态系统，为大数据处理提供了强有力的支持。TDW目前维护了上千台的Spark集群，支持了公司多个业务的挖掘分析和实时计算类任务，我们会在易用性和稳定性方面进行进一步的改进和优化，构建强大的大数据处理平台，给业务提供更有力的支持。

声明： 此文观点不代表本站立场；转载须要保留原文链接；版权疑问请联系我们。