个性化推荐调优：重写spark推荐api

fanmingchun 2018-02-07 原文

最近用spark的mlib模块中的协同过滤库做个性化推荐。spark里面用的是als算法，本质上是矩阵分解svd降维，把一个M*N的用户商品评分矩阵分解为M*K的userFeature(用户特征矩阵)和K*N的productFeature(商品特征矩阵)，由于K远小于N和M，存储和计算获得相应的优化。

这样对于一个用户a，推荐100个商品怎么做呢？取a的特征向量（1*K）和productFeature相乘得到1*M的结果向量，向量中的值代表该商品和用户a的相关度，取结果向量中前100的商品推荐给用户。

过程很简单，但是当M和N非常大呢？假设M为千万级，N为百万级，推荐一个商品需要KN+N*logN，用spark提供的单用户推荐api大约需要500ms，那么对于1000万用户，就需要500万秒，大约50几天。spark考虑到这种场景，所以提供了一次性推荐所有用户的api：recommendProductsForUsers。这个方法速度挺快，但内部采用userFeature和productFeature笛卡尔积的方法，这样产生了大量的shuffle，需要大量内存。用户量增加的时候，经常因为内存不够OOM挂掉，很不稳定。

优化势在必行，我们的目标是稳定和可扩展。分析一下整个计算过程，最大的问题就是用户量巨大且不稳定，一次性全量用户推荐需要大量内存和计算。随用户量动态调整节点数目和内存的方案，听上去很酷炫，但是调整的依据和公式又在哪呢。

简单的方案才是最好的方案，如下图。换个思路，不要一次性全量推荐了，每次推荐一部分固定数量（比如500万）的用户，切成几批，最后把结果merge起来。固定数量的用户，我们可以测出需要多少内存和节点，这样不需要扩展节点。如果用户量增加，只需要切的批次增加，多算几次，每次计算依然按照固定数量来推荐。

对于离线计算来说，多几个小时的计算时间不是问题，如果用户数量增长到推荐速度确实不够的时候，可以通过增大固定数量来解决（这种情况出现的概率很小，或者几个月后才会出现，不影响可行性）。这样就达到了我们的目的：稳定输出和可扩展。

由于spark没有这样的接口，所以只有自己写了。spark是用scala写的，深入源码用python就不行了，正好顺便把scala学了。重写过程主要是，把recommendProductsForUsers方法中的全量推荐代码复制出来稍加修改，变成自己的推荐方法，然后推荐的时候把userFeature分块去调用重写的推荐方法就可以了。

主要的收获是第一次通过修改开源代码去解决实际生产问题，黑盒变成了白盒。