ALS（alternating least squares）是一种基础的推荐算法，相对于普通的协同过滤等方法，它不仅能通过降维增加模型的泛化能力，也方便加入其他建模因素（如数据偏差、时间、隐反馈等），大大提升了模型的灵活性。正因为此，ALS算法在Netflix推荐大赛中脱颖而出，在我们具体的工程实践中，也具有非常不错的表现。接下来，从如下几个方面和大家一起学习：ALS算法模型、spark ALS源码理解， ALS推荐实践。如描述有误，欢迎大家指正。

ALS算法模型

为什么要用ALS模型

相对于其他模型，ALS模型优势如下：

相对于基于内容的推荐，ALS属于协同过滤大家族【1】【12】（也有人认为ALS 基于矩阵分解技术，不属于协同过滤范畴【2】），直接跟进用户行为信息进行建模，不需要获取user和item的内容信息（很多情况下这些内容信息并不是很好获取，但是相对基于内容的推荐，ALS存在冷启动问题）
相对于传统的协同过滤推荐方法（user based、item based）， ALS算法属于factor model, 通过将数据从原始空间映射到更低维度空间，去除噪声信息，利用更主要的语义信息对问题建模，能获得更好的推荐效果。
相对于svd分解模型而言，两种模型都属于 factor model, 但svd分解模型更倾向于解决矩阵元素没有缺失的情况，而通过一定的方式去填充矩阵不仅需要额外的填充成本，填充本身可能影响了数据的真实性。因此，直接对已知元素进行建模，是一种不错的思路。如【1，3-6】，直接对rating矩阵已知元素$r_{ui}$进行建模:

$\sum_{u,i\in\mathbb K} (r_{ui} -
p_u^Tq_i)^2 + \lambda(p_u^Tp_u+q_i^Tq_i)$ （1）

针对所建模型1可以用SGD或ALS 两种算法求解。其中sgd方法相对比较简单，但是当我们要建模的矩阵中已知元素较多时（如隐反馈），采用sgd在每次迭代都要求解所有元素，其时间复杂度是非常大的。ALS算法在求解某个user （或item）向量时，不依赖其他任何user（item）向量，这个性质使得ALS算法在每次迭代过程中方便并行化求解，在解决大规模矩阵分解问题时非常具有优势。

ALS模型有什么缺点

相对于其它推荐算法，ALS模型具有非常明显的优势：不需要对user和item信息进行建模，能够更加灵活地对各种因素建模，方便大规模并行计算。但ALS模型在如下几方面，又有自己的局限性：

冷启动问题

包括user的冷启动和item的冷启动。由于rating矩阵的构建，依赖user的显式和隐式反馈信息，对于新的user和item，或者没有相关行为的user或item, 导致无法构建rating矩阵，或者rating矩阵构建不合理。

基于内容的推荐能较好地解决冷启动相关问题。如【8】在解决用户的冷启动问题时，首先根据用户之间社交的亲密度，对用户进行聚类，利用相同群体的用户画像来建模自身兴趣。同时，对于item的冷启动问题，可利用item本身对一些关键词、类目、内容等相关信息进行建模。【9】为了提高用户兴趣的准确率和覆盖率，在对用户兴趣建模对时候，将用户兴趣进行更具体的分类（如消费兴趣、生产兴趣、具体的每个行为兴趣等），并针对具体的业务，采用线性回归的方法对各种兴趣利用线性回归的方式进行加权求和，提升用户兴趣准确率和覆盖率。
用户临时兴趣

用户的兴趣是在不断变换的。对于相对较稳定的兴趣，ALS算法可以通过引入时间因素进行建模，如公式4。但对于临时的兴趣变换，ALS算法是无法捕获的。

一种简单且有效的方法

将item划分为多个类别，每个类别对应一种兴趣。用户每次点击某个类别的item之后，认为该用户存在一种临时兴趣，通过动态增加相应类别的比例的item，迎合用户当前的消费需求。该方法的难点在于如何调整比例，才能让用户感到有很多自己喜欢的item, 同时又不会让用户感觉内容的单调。

淘宝的一些实践

为了有效获取用户的即时兴趣，给用户推荐最合适的产品，淘宝进行了比较多的实践【10】，分别如下所述。
- GBDT+FTRL模型
  
  由于GBDT模型比较擅长挖具有区分度的特征，其使用GBDT模型进行特征挖掘，将得到的特征输送给FTRL进行在线学习。输送给GBDT的特征包括两部分：一部分用户基础行为的次数、CTR等；另一部分是来自match粗选阶段的的特征，该部分特征来自不同的粗选模型输出.
- Wide & Deep Learning模型
  
  借鉴google论文思想【11】，利用wide模型 + deep模型 + LR，其中wide子结构通过特征交叉学习特征间的共现，deep子结构则输入具有泛化能力的离散特征和连续特征，wide模型和deep模型学习到的结果，再利用LR模型预测相应的得分。
- Adaptive-Online-Learning
  
  保留每一时刻学习到的模型，根据业务指标，得到每个模型等权重信息，融合出最优的结果。该方法能够比较好地综合利用用户长期喝短期兴趣。
- Reinforcement Learning
  
  该方法思想是通过定义每个步骤的奖励，当用户每次到来的时候，根据用户的累积奖励值，进行个性化推荐。
腾讯的LSTM实践

为解决音乐的推荐问题，腾讯采用的是LSTM深度学习方法【12】，将用户听的歌曲序列，抽取特征输入到LSTM网络进行训练。为防止有些用户对应的歌曲序列较短问题，其对这些数据的训练采用特殊处理，相关数据缺失的序列不进行状态更新。同时，为加快训练速度，将每次权值的训练过程通过矩阵的方式实现并发计算。另外，为降低soft max过程时间复杂度，采用Hierarchical softmax过程替代普通的softmax。

ALS模型是什么

基本概念

ALS模型属于隐语义模型，通过对用户行为矩阵R进行矩阵分解，得到user factor向量矩阵P、item factor向量矩阵Q.

$R = P^T Q$ 。其中R、$P^T$、$Q^T$矩阵的定义如表1-表3所示。

潜在语义空间对应的各个factor代表不同的属性信息，user向量描述了user对各种属性的喜好程度，item向量描述了item所具备的各种属性强度，二者在潜在语义空间的相似度描述了user对item的喜好程度,在进行推荐时，根据该喜好程度计算推荐结果。

表1: rating矩阵R

	item1	item2	item3	item4
user1	$r_{11}$	$r_{12}$	$r_{13}$	$r_{14}$
user2	$r_{21}$	$r_{22}$	$r_{23}$	$r_{24}$
user3	$r_{31}$	$r_{32}$	$r_{33}$	$r_{34}$
user4	$r_{41}$	$r_{42}$	$r_{43}$	$r_{44}$
user5	$r_{51}$	$r_{52}$	$r_{53}$	$r_{54}$

表2：user矩阵$P^T$

	factor1	factor2	factor3
user1	$p_{11}$	$p_{12}$	$p_{13}$
user2	$p_{21}$	$p_{22}$	$p_{23}$
user3	$p_{31}$	$p_{32}$	$p_{33}$
user4	$p_{41}$	$p_{42}$	$p_{43}$
user5	$p_{51}$	$p_{52}$	$p_{53}$

表3:item矩阵$Q^T$

	factor1	factor2	factor3
item1	$q_{11}$	$q_{12}$	$q_{13}$
item2	$q_{21}$	$q_{22}$	$q_{23}$
item3	$q_{31}$	$q_{32}$	$q_{33}$
item4	$q_{41}$	$q_{42}$	$q_{43}$

目标函数

$MIN_{PQ} \sum_{u,i\in\mathbb K} {(r_{ui} -
p_u^Tq_i）}^2 + \lambda(p_u^Tp_u+q_i^Tq_i)$ (2)

其中${(r_{ui} - p_u^Tq_i）}^2$ 目的在于最小化分解误差，$\lambda(p_u^Tp_u+q_i^Tq_i)$ 为正则项。

目标函数求解

由于目标函数中$p_u, q_i$都是未知变量，该问题是非凸的。当我们固定其中一个变量，解另外一个变量时，问题则变成凸问题，这是ALS求解的主要思想。在实际求解过程中分为如下几个步骤：

随机初始化所有的变量$p_u, q_i$。
固定所有的$q_i$变量，求得$q_i$变量为当前值时$p_u$的最优值。
固定所有的$p_u$变量，求得$p_u$变量为当前值时$q_i$的最优值。
如果满足终止条件，则终止。否则，迭代执行2，3两步。

通过不断执行步骤2和步骤3，使得误差越来越小，直到收敛或达到指定次数而终止。通过可导函数性质我们知道，当对变量求导结果等于0当时候，函数可以取得极值。具体到公式2，固定一个变量，对另一变量求导结果等于0时，可以达到极小值。

我们令$L = \sum_{u,i\in\mathbb K} {(r_{ui} - p_u^Tq_i）}^2 + \lambda(p_u^Tp_u+q_i^Tq_i)$

固定所有$q_i$, 对$p_u$求导

$-\frac{\alpha L}{2\alpha p_{uk}} = \sum_{i} {q_{ik}(r_{ui} - p_u^Tq_i）} - \lambda p_{uk} = 0$

=> $\sum_{i} {q_{i}(r_{ui} - p_u^Tq_i）} - \lambda p_{u} = 0$

=> $(\sum_{i} {q_i q_i^T} + \lambda E) p_u = \sum_{i}q_i r_{ui}$

=> $p_u = (\sum_{i} {q_i q_i^T} + \lambda E)^{-1}\sum_{i}q_i r_{ui}$

=> $ p_u = (Q_{u,i\in\mathbb K} Q_{u,i\in\mathbb K}^T + \lambda E)^{-1}Q_{u,i\in\mathbb K}R_{u,i\in\mathbb K}^T$

其中，$q_{u,i\in\mathbb K}$ 表示和user $u$有行为关联的item对应的向量矩阵，$r_{u,i\in\mathbb K}^T$表示和user $u$有行为关联的item对应rating元素构成的向量的转置。

更加灵活的ALS建模

相对于传统的协同协同过滤方法，ALS能更好的考虑其他因素，如数据偏差、时间等

引入数据偏差

user偏差：不同的用户，可能具有不同的评分标准。如用户在给item打分时，有的用户可能可能更倾向于给所有item打高分，而有的挑剔用户会给所有item打分偏低

item偏差：有的热门item可能所有用户都会倾向于打高分，而有的item可能本身大多数人会倾向于打低分

考虑use和item偏差的ALS建模：
$MIN_{PQB} \sum_{u,i\in\mathbb K} {(r_{ui} - u - b_u - b_i-
p_u^Tq_i）}^2 + \lambda(p_u^Tp_u+q_i^Tq_i+b_u^2+b_i^2)$ (3)
引入时间因素

用户偏好、rating矩阵，都可能随时间变化，item对应的属性不随时间变化，因此可进行如下建模
$MIN_{PQB} \sum_{u,i\in\mathbb K} {(r_{ui}（t） - u - b_u(t) - b_i(t)-
p_u(t)^Tq_i）}^2 + \lambda(p_u(t)^Tp_u(t)+q_i^Tq_i+b_u(t)^2+b_i(t)^2)$ (4)
引入隐反馈数据因素

很多时候，并没有用户对item明确的打分数据，此时可通过搜集用户隐反馈数据（浏览、点击、点赞等），进行隐反馈建模。有一点需要注意，此时不只是对$r_{ui}$大于0对用户行为建模，而是所有$r_{ui}$元素建模。模型如公式5所示：

$MIN_{PQB} \sum_{u,i} {c_{ui}(p_{ui} - u - b_u - b_i-
p_u^Tq_i）}^2 + \lambda(p_u^Tp_u+q_i^Tq_i+b_u^2+b_i^2)$ (5)

$p_{ui}$ 表示user u是否有相关行为表示喜欢item i, $c_{ui}$描述user u 对item i的喜欢程度，其定义如公式6和公式7所示

$
p_{ui} =
\begin{cases}
1, & r_{ui}>0\\
0, & r_{ui}=0
\end{cases}
$（6）

$c_{ui} = 1 + \alpha r_{ui}$（7）

spark ALS源码理解

为加深对ALS算法的理解，该部分主要分析spark mllib中ALS源码的实现，大体上分为2部分：ALS模型训练、ALS模型推荐

ALS 模型训练

ALS 伴生类

ALS 伴生对象提供外部调用 ALS模型训练的入口。通过传入相关参数，返回训练好的模型对象MatrixFactorizationModel。

object ALS {
  def train(
      ratings: RDD[Rating], //rating元素 （user, item, rate）
      rank: Int, //隐语义个数
      iterations: Int, //迭代次数
      lambda: Double, //正则惩罚项
      blocks: Int, //数据block个数
      seed: Long //随机数种子
    ): MatrixFactorizationModel = {
    new ALS(blocks, blocks, rank, iterations, lambda, fALSe, 1.0, seed).run(ratings)
  }

  def trainImplicit(
      ratings: RDD[Rating], // rating元素 （user, item, rate）
      rank: Int, //隐语义个数
      iterations: Int, //迭代次数
      lambda: Double, //正则惩罚项
      blocks: Int, //数据block个数
      alpha: Double //计算$c_ui$时用的alpha参数
    ): MatrixFactorizationModel = {
    new ALS(blocks, blocks, rank, iterations, lambda, true, alpha).run(ratings)
  }
   //另外还有一些其他接口，因最终都通过调用上面2个函数，此处将其省略
}

ALS 私有类

定义了ALS类对应的各个参数，以及各个参数的设定方法。并定义了run方法供伴随类进行调用，该方法返回训练结果MatrixFactorizationModel给ALS伴随类。

class ALS private (
    private var numUserBlocks: Int, //用户数据block个数
    private var numProductBlocks: Int, //item数据block个数
    private var rank: Int, //隐语义个数
    private var iterations: Int, //迭代次数
    private var lambda: Double, //正则惩罚项
    private var implicitPrefs: Boolean, //是否使用隐反馈模型
    private var alpha: Double, //计算$c_ui$时用的alpha参数
    private var seed: Long = System.nanoTime() //随机数种子,默认为当前时间戳
  ) extends Serializable with Logging {

  //设置block个数
  def setBlocks(numBlocks: Int): this.type = {
    this.numUserBlocks = numBlocks
    this.numProductBlocks = numBlocks
    this
  }
  
  // 另外对其他参数变量也有相关函数实现，因基本都是赋值操作，此处将其省略
  
  //run方法，通过输入rating数据，完成训练兵返回结果MatrixFactorizationModel
  def run(ratings: RDD[Rating]): MatrixFactorizationModel = {
    require(!ratings.isEmpty(), s"No ratings available from $ratings")

    val sc = ratings.context
    //设置user block个数
    val numUserBlocks = if (this.numUserBlocks == -1) {
      math.max(sc.defaultParallelism, ratings.partitions.length / 2)
    } else {
      this.numUserBlocks
    }
    //设置item block个数
    val numProductBlocks = if (this.numProductBlocks == -1) {
      math.max(sc.defaultParallelism, ratings.partitions.length / 2)
    } else {
      this.numProductBlocks
    }
    //调用NewALS.train方法完成矩阵分解，生成user factor和item factor向量,该方法是整个ALS算法的核心实现
    val (floatUserFactors, floatProdFactors) = NewALS.train[Int](
      ratings = ratings.map(r => NewALS.Rating(r.user, r.product, r.rating.toFloat)),
      rank = rank,
      numUserBlocks = numUserBlocks,
      numItemBlocks = numProductBlocks,
      maxIter = iterations,
      regParam = lambda,
      implicitPrefs = implicitPrefs,
      alpha = alpha,
      nonnegative = nonnegative,
      intermediateRDDStorageLevel = intermediateRDDStorageLevel,
      finalRDDStorageLevel = StorageLevel.NONE,
      checkpointInterval = checkpointInterval,
      seed = seed)
   
    val userFactors = floatUserFactors
      .mapValues(_.map(_.toDouble))
      .setName("users")
      .persist(finalRDDStorageLevel)
    val prodFactors = floatProdFactors
      .mapValues(_.map(_.toDouble))
      .setName("products")
      .persist(finalRDDStorageLevel)
    if (finalRDDStorageLevel != StorageLevel.NONE) {
      userFactors.count()
      prodFactors.count()
    }
    //生成和返回ALS模型 MatrixFactorizationModel
    new MatrixFactorizationModel(rank, userFactors, prodFactors)
  }
}

NewALS.train方法

被ALS私有类的run方法调用，用于计算user factor和item factor向量。

def train[ID: ClassTag]( // scalastyle:ignore
      ratings: RDD[Rating[ID]],
      rank: Int = 10,
      numUserBlocks: Int = 10,
      numItemBlocks: Int = 10,
      maxIter: Int = 10,
      regParam: Double = 1.0,
      implicitPrefs: Boolean = fALSe,
      alpha: Double = 1.0,
      nonnegative: Boolean = fALSe,
      intermediateRDDStorageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK,
      finalRDDStorageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK,
      checkpointInterval: Int = 10,
      seed: Long = 0L)(
      implicit ord: Ordering[ID]): (RDD[(ID, Array[Float])], RDD[(ID, Array[Float])]) = {
    require(!ratings.isEmpty(), s"No ratings available from $ratings")
    require(intermediateRDDStorageLevel != StorageLevel.NONE,
      "ALS is not designed to run without persisting intermediate RDDs.")
    val sc = ratings.sparkContext
    //根据block个数，构建哈稀器。
    val userPart = new ALSPartitioner(numUserBlocks)
    val itemPart = new ALSPartitioner(numItemBlocks)
    //构建索引编码器，根据block编号和block内索引进行编码，同时可将编码后结果快速解码为block编号和block内索引号。具体实现是通过block个数，确定block编码需要的二进制位数，以及block内索引位数，通过这些位数利用逻辑操作即可实现编码和解码

    val userLocalIndexEncoder = new LocalIndexEncoder(userPart.numPartitions)
    val itemLocalIndexEncoder = new LocalIndexEncoder(itemPart.numPartitions)
    //构建求解器
    val solver = if (nonnegative) new NNLSSolver else new CholeskySolver
    //对rating矩阵进行分块，得到((user_blockID, item_blockID),rating(user, item, rating))

    val blockRatings = partitionRatings(ratings, userPart, itemPart)
      .persist(intermediateRDDStorageLevel)
    //构建user inblock和outblock数据，inblock数据记录每个user对应的所有item的地址，及对应rating信息。 outblock记录当前block的哪些user数据会被哪些block用上
    val (userInBlocks, userOutBlocks) =
      makeBlocks("user", blockRatings, userPart, itemPart, intermediateRDDStorageLevel)
    // materialize blockRatings and user blocks
    userOutBlocks.count()
    //交换blockrating中的user, item数据，用于构造item的inblcok和outblock信息
    val swappedBlockRatings = blockRatings.map {
      case ((userBlockId, itemBlockId), RatingBlock(userIds, itemIds, localRatings)) =>
        ((itemBlockId, userBlockId), RatingBlock(itemIds, userIds, localRatings))
    }
    val (itemInBlocks, itemOutBlocks) =
      makeBlocks("item", swappedBlockRatings, itemPart, userPart, intermediateRDDStorageLevel)
    // materialize item blocks
    itemOutBlocks.count()
    val seedGen = new XORShiftRandom(seed)
    //随机初始化user factor和item factor
    var userFactors = initialize(userInBlocks, rank, seedGen.nextLong())
    var itemFactors = initialize(itemInBlocks, rank, seedGen.nextLong())
    var previousCheckpointFile: Option[String] = None
    val shouldCheckpoint: Int => Boolean = (iter) =>
      sc.checkpointDir.isDefined && checkpointInterval != -1 && (iter % checkpointInterval == 0)
    val deletePreviousCheckpointFile: () => Unit = () =>
      previousCheckpointFile.foreach { file =>
        try {
          val checkpointFile = new Path(file)
          checkpointFile.getFileSystem(sc.hadoopConfiguration).delete(checkpointFile, true)
        } catch {
          case e: IOException =>
            logWarning(s"Cannot delete checkpoint file $file:", e)
        }
      }
    //针对隐反馈，迭代求解
    if (implicitPrefs) {
      for (iter <- 1 to maxIter) {  //迭代总次数maxIter
        userFactors.setName(s"userFactors-$iter").persist(intermediateRDDStorageLevel)
        val previousItemFactors = itemFactors
        //固定user factor，优化item factor
        itemFactors = computeFactors(userFactors, userOutBlocks, itemInBlocks, rank, regParam,
          userLocalIndexEncoder, implicitPrefs, alpha, solver)
        previousItemFactors.unpersist()
        itemFactors.setName(s"itemFactors-$iter").persist(intermediateRDDStorageLevel)
        // TODO: Generalize PeriodicGraphCheckpointer and use it here.
        val deps = itemFactors.dependencies
        if (shouldCheckpoint(iter)) {
          itemFactors.checkpoint() // itemFactors gets materialized in computeFactors
        }
        val previousUserFactors = userFactors
        //根据item factore, 优化user factor
        userFactors = computeFactors(itemFactors, itemOutBlocks, userInBlocks, rank, regParam,
          itemLocalIndexEncoder, implicitPrefs, alpha, solver)
        if (shouldCheckpoint(iter)) {
          ALS.cleanShuffleDependencies(sc, deps)
          deletePreviousCheckpointFile()
          previousCheckpointFile = itemFactors.getCheckpointFile
        }
        previousUserFactors.unpersist()
      }
    } else { //针对显示反馈，迭代求解
      for (iter <- 0 until maxIter) { //迭代总次数maxIter
        //固定user factor，优化item factor
        itemFactors = computeFactors(userFactors, userOutBlocks, itemInBlocks, rank, regParam,
          userLocalIndexEncoder, solver = solver)
        if (shouldCheckpoint(iter)) {
          val deps = itemFactors.dependencies
          itemFactors.checkpoint()
          itemFactors.count() // checkpoint item factors and cut lineage
          ALS.cleanShuffleDependencies(sc, deps)
          deletePreviousCheckpointFile()
          previousCheckpointFile = itemFactors.getCheckpointFile
        }
        //根据item factore, 优化user factor
        userFactors = computeFactors(itemFactors, itemOutBlocks, userInBlocks, rank, regParam,
          itemLocalIndexEncoder, solver = solver)
      }
    }
    //将user id 和 factor拼接在一起
    val userIdAndFactors = userInBlocks
      .mapValues(_.srcIds)
      .join(userFactors)
      .mapPartitions({ items =>
        items.flatMap { case (_, (ids, factors)) =>
          ids.view.zip(factors)
        }
      // Preserve the partitioning because IDs are consistent with the partitioners in userInBlocks
      // and userFactors.
      }, preservesPartitioning = true)
      .setName("userFactors")
      .persist(finalRDDStorageLevel)
    //将item id 和 factor拼接在一起
    val itemIdAndFactors = itemInBlocks
      .mapValues(_.srcIds)
      .join(itemFactors)
      .mapPartitions({ items =>
        items.flatMap { case (_, (ids, factors)) =>
          ids.view.zip(factors)
        }
      }, preservesPartitioning = true)
      .setName("itemFactors")
      .persist(finalRDDStorageLevel)
    if (finalRDDStorageLevel != StorageLevel.NONE) {
      userIdAndFactors.count()
      itemFactors.unpersist()
      itemIdAndFactors.count()
      userInBlocks.unpersist()
      userOutBlocks.unpersist()
      itemInBlocks.unpersist()
      itemOutBlocks.unpersist()
      blockRatings.unpersist()
    }
    //返回user factor和item factor数据
    (userIdAndFactors, itemIdAndFactors)
  }

构建哈希器

构建哈希器，用于计算user或item id对应的block编号。

class HashPartitioner(partitions: Int) extends Partitioner {
  require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")
  //block总数
  def numPartitions: Int = partitions
  //通过求余计算block 编号
  def getPartition(key: Any): Int = key match {
    case null => 0
    case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
  }
  //判断2个哈希器是否相等
  override def equALS(other: Any): Boolean = other match {
    case h: HashPartitioner =>
      h.numPartitions == numPartitions
    case _ =>
      fALSe
  }
  override def hashCode: Int = numPartitions
}

构建地址编码解码器

构建地址编码解码器，根据block编号和block内索引对地址进行编码，同时可将编码后地址解码为block编号和block内索引号。具体实现是通过block个数确定block编码需要的二进制位数，以及block内索引位数，通过这些位数利用逻辑操作即可实现地址的编码和解码。

private[recommendation] class LocalIndexEncoder(numBlocks: Int) extends Serializable {

    require(numBlocks > 0, s"numBlocks must be positive but found $numBlocks.")
    //block内部索引使用的二进制位数
    private[this] final val numLocalIndexBits =
      math.min(java.lang.Integer.numberOfLeadingZeros(numBlocks - 1), 31)
    private[this] final val localIndexMask = (1 << numLocalIndexBits) - 1
    
    //根据block编号和block内索引值，对地址编码
    def encode(blockId: Int, localIndex: Int): Int = {
      require(blockId < numBlocks)
      require((localIndex & ~localIndexMask) == 0)
      (blockId << numLocalIndexBits) | localIndex
    }

    //根据编码后地址，得到block编号
    def blockId(encoded: Int): Int = {
      encoded >>> numLocalIndexBits
    }

    //根据编码地址，得到block内部索引
    def localIndex(encoded: Int): Int = {
      encoded & localIndexMask
    }
  }

partition rating

格式化rating数据，将rating数据分块，根据user和product的id哈希后的结果，得到对应的块索引。最终返回（src_block_id, dst_block_id）(src_id数组，dst_id数组，rating数组)

private def partitionRatings[ID: ClassTag](
      ratings: RDD[Rating[ID]],
      srcPart: Partitioner,
      dstPart: Partitioner): RDD[((Int, Int), RatingBlock[ID])] = {
    //获得总block数
    val numPartitions = srcPart.numPartitions * dstPart.numPartitions
    //在rating的每个分区，计算每个rating元素对应的src_block_id和dst_block_id, 并放到对应的块索引中。然后，对所有分区的元素按照块索引进行聚合，并返回聚合结果
    ratings.mapPartitions { iter =>
      //生成numPartitions个一维数组，存储对应block的rating记录
      val builders = Array.fill(numPartitions)(new RatingBlockBuilder[ID])
      iter.flatMap { r =>
        val srcBlockId = srcPart.getPartition(r.user) //user block id
        val dstBlockId = dstPart.getPartition(r.item) //item block id
        val idx = srcBlockId + srcPart.numPartitions * dstBlockId //数组索引计算
        //将对应的rating元素放在builders对应元素中
        val builder = builders(idx) 
        builder.add(r) 
        if (builder.size >= 2048) { // 2048 * (3 * 4) = 24k
          //如果某个block内数据量较多，直接得到结果
          builders(idx) = new RatingBlockBuilder
          Iterator.single(((srcBlockId, dstBlockId), builder.build()))
        } else {
          Iterator.empty
        }
      } ++ {
        //对builders数组内元素，计算对应的src_block_id和dst_block_id,并将对应rating数据放在其中
        builders.view.zipWithIndex.filter(_._1.size > 0).map { case (block, idx) =>
          val srcBlockId = idx % srcPart.numPartitions
          val dstBlockId = idx / srcPart.numPartitions
          ((srcBlockId, dstBlockId), block.build())
        }
      }
    }.groupByKey().mapValues { blocks =>
      //对不同分区计算出的的rating元素进行聚合
      val builder = new RatingBlockBuilder[ID]
      blocks.foreach(builder.merge)
      builder.build() //value为 （src_id数组，dst_id数组，对应的rating数组）
    }.setName("ratingBlocks")
  }

构造in_block, 和out_block

在分布式计算中，不同节点的通信是影响程序效率重要原因，通过合理的设计分区，使得不同节点交换数据尽量少，可以有效的提升运行效率。

由上述章节中对目标函数求解推导，可以得知，每个用户向量的计算依赖于所有和它关联的item向量。如果不做任何优化，则每次优化user向量时，所有user向量的计算，都需要从其他节点得到对应item向量。如果节点A上有多个user和节点B上的某一item关联，则节点B需要向节点A传输多次item向量数据，实际上这是不必要的。优化的思路是，通过合理的分区，提前计算好所有节点需要从其它节点获取的item向量数据，将其缓存在本地，计算每个user向量时，直接从本地读取，可以大大减少需要传输的数据量，提升程序执行的效率。

在源码中，通过out block缓存当前节点需要向其它节点传输的数据， in block用于缓存当前节点需要的数据索引。当其他节点信息传输到本地时，通过读取in block内索引信息，来从本地获取其它节点传过来的数据。更加详细的描述可参考【7】

in block 结构：（block_id, Inblock(src_id数组, src_ptr, dst_id地址数组， rating数组）)
out block结构：（block_id， array[array[int]]）（二维数组存储发往每个block的src_id索引）

private def makeBlocks[ID: ClassTag](
      prefix: String,
      ratingBlocks: RDD[((Int, Int), RatingBlock[ID])],
      srcPart: Partitioner,
      dstPart: Partitioner,
      storageLevel: StorageLevel)(
      implicit srcOrd: Ordering[ID]): (RDD[(Int, InBlock[ID])], RDD[(Int, OutBlock)]) = {
    //根据ratingBlocks.map计算inBlocks
    val inBlocks = ratingBlocks.map {
      case ((srcBlockId, dstBlockId), RatingBlock(srcIds, dstIds, ratings)) =>
        val start = System.nanoTime()
        //dst id去重复
        val dstIdSet = new OpenHashSet[ID](1 << 20) 
        dstIds.foreach(dstIdSet.add)  
        //dst id 去重结果进行排序
        val sortedDstIds = new Array[ID](dstIdSet.size)
        var i = 0
        var pos = dstIdSet.nextPos(0)
        while (pos != -1) {
          sortedDstIds(i) = dstIdSet.getValue(pos)
          pos = dstIdSet.nextPos(pos + 1)
          i += 1
        }
        assert(i == dstIdSet.size)
        Sorting.quickSort(sortedDstIds)
        //得到dst id 对应的去重和排序后的索引值
        val dstIdToLocalIndex = new OpenHashMap[ID, Int](sortedDstIds.length)
        i = 0
        while (i < sortedDstIds.length) {
          dstIdToLocalIndex.update(sortedDstIds(i), i)
          i += 1
        }
        logDebug(
          "Converting to local indices took " + (System.nanoTime() - start) / 1e9 + " seconds.")
        val dstLocalIndices = dstIds.map(dstIdToLocalIndex.apply)
        (srcBlockId, (dstBlockId, srcIds, dstLocalIndices, ratings))
    }.groupByKey(new ALSPartitioner(srcPart.numPartitions)) //根据src block id进行聚合
      .mapValues { iter =>
        val builder =
          new UncompressedInBlockBuilder[ID](new LocalIndexEncoder(dstPart.numPartitions))
        //将dstBlockId和dstLocalIndices编码，并汇总数据
        iter.foreach { case (dstBlockId, srcIds, dstLocalIndices, ratings) =>
          builder.add(dstBlockId, srcIds, dstLocalIndices, ratings)
        }
        //对结果进行压缩存储，结果格式为（uniqueSrcId数组, dstPtrs数组, dstEncodedIndices数组, ratings数组）
        builder.build().compress()
      }.setName(prefix + "InBlocks")
      .persist(storageLevel)
        
    //根据inBlocks计算outBlocks
    val outBlocks = inBlocks.mapValues { case InBlock(srcIds, dstPtrs, dstEncodedIndices, _) =>
      //构造编码器
      val encoder = new LocalIndexEncoder(dstPart.numPartitions)
      //定义ArrayBuilder数组，存储发往每个out block的 src id信息
      val activeIds = Array.fill(dstPart.numPartitions)(mutable.ArrayBuilder.make[Int])
      var i = 0
      val seen = new Array[Boolean](dstPart.numPartitions)
      //依次计算当前src id是否发往每一个block id
      while (i < srcIds.length) {
        var j = dstPtrs(i)
        ju.Arrays.fill(seen, fALSe)
        while (j < dstPtrs(i + 1)) {
          val dstBlockId = encoder.blockId(dstEncodedIndices(j))
          if (!seen(dstBlockId)) {
            activeIds(dstBlockId) += i // add the local index in this out-block
            seen(dstBlockId) = true
          }
          j += 1
        }
        i += 1
      }
      activeIds.map { x =>
        x.result()
      }
    }.setName(prefix + "OutBlocks")
      .persist(storageLevel)
    (inBlocks, outBlocks)  //返回结果
  }

inblock compress

对inblock 中间结果压缩存储，返回结果格式为（uniqueSrcId数组, dstPtrs数组, dstEncodedIndices数组, ratings数组）

def compress(): InBlock[ID] = {
  val sz = length
  assert(sz > 0, "Empty in-link block should not exist.")
  sort()
  val uniqueSrcIdsBuilder = mutable.ArrayBuilder.make[ID]
  val dstCountsBuilder = mutable.ArrayBuilder.make[Int]
  var preSrcId = srcIds(0)
  uniqueSrcIdsBuilder += preSrcId
  var curCount = 1
  var i = 1
  var j = 0
  //得到去重后的src id数组， 以及每个src id的数量
  while (i < sz) {
    val srcId = srcIds(i)
    if (srcId != preSrcId) {
      uniqueSrcIdsBuilder += srcId
      dstCountsBuilder += curCount
      preSrcId = srcId
      j += 1
      curCount = 0
    }
    curCount += 1
    i += 1
  }
  dstCountsBuilder += curCount
  val uniqueSrcIds = uniqueSrcIdsBuilder.result()
  val numUniqueSrdIds = uniqueSrcIds.length
  val dstCounts = dstCountsBuilder.result()
  val dstPtrs = new Array[Int](numUniqueSrdIds + 1)
  var sum = 0
  //将src id和dst id关系通过dstPtrs进行压缩存储
  i = 0
  while (i < numUniqueSrdIds) {
    sum += dstCounts(i)
    i += 1
    dstPtrs(i) = sum
  }
  InBlock(uniqueSrcIds, dstPtrs, dstEncodedIndices, ratings)
}

computeFactor

根据srcFactorBlocks、srcOutBlocks、dstInBlocks, 计算dstFactorBlocks

private def computeFactors[ID](
    srcFactorBlocks: RDD[(Int, FactorBlock)],
    srcOutBlocks: RDD[(Int, OutBlock)],
    dstInBlocks: RDD[(Int, InBlock[ID])],
    rank: Int,
    regParam: Double,
    srcEncoder: LocalIndexEncoder,
    implicitPrefs: Boolean = fALSe,
    alpha: Double = 1.0,
    solver: LeastSquaresNESolver): RDD[(Int, FactorBlock)] = {
  val numSrcBlocks = srcFactorBlocks.partitions.length  //src block数量
  val YtY = if (implicitPrefs) Some(computeYtY(srcFactorBlocks, rank)) else None
  //根据srcOut，得到每个dstBlock对应的srcBlockID 和srcFactor数组
  val srcOut = srcOutBlocks.join(srcFactorBlocks).flatMap {
    case (srcBlockId, (srcOutBlock, srcFactors)) =>
      
      srcOutBlock.view.zipWithIndex.map { case (activeIndices, dstBlockId) =>
        (dstBlockId, (srcBlockId, activeIndices.map(idx => srcFactors(idx))))
      }
  }
  //根据dstBlockId 对srcBlockID, array[srcFactor]进行聚合
  val merged = srcOut.groupByKey(new ALSPartitioner(dstInBlocks.partitions.length))
  //对每个dstBlockID, 计算其中每个dstID对应的隐语义向量
  dstInBlocks.join(merged).mapValues {
    case (InBlock(dstIds, srcPtrs, srcEncodedIndices, ratings), srcFactors) =>
      //得到每个block对应的src factor向量集合
val sortedSrcFactors = new Array[FactorBlock](numSrcBlocks)
      srcFactors.foreach { case (srcBlockId, factors) =>
        sortedSrcFactors(srcBlockId) = factors
      }
      //对每个dstID, 获取对应的srcFactor及对应rating, 计算该dstID对应的隐语义向量
      val dstFactors = new Array[Array[Float]](dstIds.length)
      var j = 0
      val ls = new NormalEquation(rank)
      while (j < dstIds.length) {
        ls.reset()
        if (implicitPrefs) {
          ls.merge(YtY.get)
        }
        var i = srcPtrs(j)
        var numExplicits = 0
        while (i < srcPtrs(j + 1)) { //依次得到每个srcFactor及rating值
          val encoded = srcEncodedIndices(i)
          val blockId = srcEncoder.blockId(encoded)
          val localIndex = srcEncoder.localIndex(encoded)
          //sortedSrcFactors通过blockId和localIndex进行索引，得到需要的factor向量。之前这里困惑挺久，一直感觉从srcOut传过来的factor向量只是一个子集，通过localIndex访问不正确，实际上这里的localIndex和srcOut那里存储的localindex是不需要对应的。因为同一个src id 本身的src local index不等于其它block对应的 dst localindex
          val srcFactor = sortedSrcFactors(blockId)(localIndex)
          val rating = ratings(i)
          if (implicitPrefs) {
            // Extension to the original paper to handle b < 0. confidence is a function of |b|
            // instead so that it is never negative. c1 is confidence - 1.0.
            val c1 = alpha * math.abs(rating)
            // For rating <= 0, the corresponding preference is 0. So the term below is only added
            // for rating > 0. Because YtY is already added, we need to adjust the scaling here.
            if (rating > 0) {
              numExplicits += 1
              ls.add(srcFactor, (c1 + 1.0) / c1, c1)
            }
          } else {
            ls.add(srcFactor, rating)
            numExplicits += 1
          }
          i += 1
        }
        // Weight lambda by the number of explicit ratings based on the ALS-WR paper.
        dstFactors(j) = solver.solve(ls, numExplicits * regParam)
        j += 1
      }
      dstFactors
  }
}

ALS 模型推荐

模型参数

1
2
3

val rank: Int,      //隐语义个数
val userFeatures: RDD[(Int, Array[Double])], //user factor数组, 存储user id 及对应的factor向量
val productFeatures: RDD[(Int, Array[Double])]) //item factor数组，存储item id及对应的factor向量

对所有用户进行推荐

调用recommendForAll函数，首先对user向量和item向量分块并以矩阵形式存储，然后对二者做笛卡尔积，并计算每个user和每个item的得分，最终以user为key, 取topK个item及对应的得分，作为推荐结果. 计算topK时借助于小顶堆

private def recommendForAll(
    rank: Int,
    srcFeatures: RDD[(Int, Array[Double])],
    dstFeatures: RDD[(Int, Array[Double])],
    num: Int): RDD[(Int, Array[(Int, Double)])] = {
  //对user向量和item向量分块并以矩阵形式存储
  val srcBlocks = blockify(rank, srcFeatures)
  val dstBlocks = blockify(rank, dstFeatures)
  //笛卡尔积，依次对每个组合计算user对item的偏好
  val ratings = srcBlocks.cartesian(dstBlocks).flatMap {
    case ((srcIds, srcFactors), (dstIds, dstFactors)) =>
      val m = srcIds.length
      val n = dstIds.length
      val ratings = srcFactors.transpose.multiply(dstFactors)
      val output = new Array[(Int, (Int, Double))](m * n)
      var k = 0
      ratings.foreachActive { (i, j, r) =>
        output(k) = (srcIds(i), (dstIds(j), r))
        k += 1
      }
      output.toSeq
  }
  //根据user id作为key, 得到喜好分数最高的num个item
  ratings.topByKey(num)(Ordering.by(_._2))
}


// 对user向量和item向量分块并以矩阵形式存储, 结果的每个元组分别是对应的id数组和factor构成的矩阵
private def blockify(
    rank: Int,
    features: RDD[(Int, Array[Double])]): RDD[(Array[Int], DenseMatrix)] = {
  val blockSize = 4096 // TODO: tune the block size
  val blockStorage = rank * blockSize
  features.mapPartitions { iter =>
    iter.grouped(blockSize).map { grouped =>
      val ids = mutable.ArrayBuilder.make[Int]
      ids.sizeHint(blockSize)
      val factors = mutable.ArrayBuilder.make[Double]
      factors.sizeHint(blockStorage)
      var i = 0
      grouped.foreach { case (id, factor) =>
        ids += id
        factors ++= factor
        i += 1
      }
      (ids.result(), new DenseMatrix(rank, i, factors.result()))
    }
  }
}

ALS推荐实践

我们的平台是图片社交，每个用户都可以在平台上浏览图片，并进行点赞、评论等。推荐算法主要用于给用户推荐其最可能感兴趣的图片，最终提升用户体验。

离线实验

我们平台暂时无法得到用户的显式评分数据，但是可以得到用户点击、点赞、评论等相关行为信息。因此，比较适合用隐反馈矩阵分解模型。

构造数据集

数据预处理

从2周的用户行为数据中，过滤无行为用户数据，spam图片数据和spam用户数据。
构建rating元素

对预处理之后的数据，根据用户每天的图片交互行为，分别对点击、点赞和评论等分别赋予不同的权值，得到rating矩阵.
生成训练集和测试集

对于得到的rating数据，随机划分为两部分 $A:B = 7:3$，如果分别直接作为训练集和测试集是有问题的，因为$B$中的user或者item是有可能在$A$中没有出现过，这样会影响评估结果。我们采用的方法是如果B数据中某个rating元素的user或item没有在A出现，则将该元素放到$A$中用作训练集。最终$A$和新加进来的元素共同构成训练集$A^1$， $B$留下的数据 $B^1$ 作为测试集。

离线训练和评估

离线训练

利用spark mllib库，对训练集构成的rating矩阵，建立隐反馈矩阵分解模型，并完成进行矩阵分解，生成user factor和item factor。

评估

调用模型的recommendForAll函数，对测试集所有user进行item推荐，并计算召回率和准确率。根据召回率和准确率，进行参数优化。

评估指标

假定$P_i$为用户$i$的预测结果，$P$为所有的预测结果，每个结果记录格式为（user, item）， $T$为测试集,每条记录格式为（user， item）。各种指标的的计算如下：

召回率: $R= \frac{|P \bigcap T|} {|T|}$

准确率: $P= \frac{|P \bigcap T|} {|P|}$

F1: $F= \frac{2PR} {P+R} $

离散度：$\frac{1}{N^2}\sum_i\sum_j\frac{|P_i \bigcap P_j|}{|P_i \bigcup P_j|}$

除了上述指标之外，我们还对用户连续多天推荐结果的差异性、用户覆盖率、图片覆盖率等指标进行评估。

在线ab测试

abtest方案：将als算法计算出的结果，定期写入到线上，作为线上的一种推荐来源。对实验组用户同时采用新策略和旧策略进行推荐，对照组用户只采用旧策略进行推荐。

从2个维度进行评估：

评估实验组和对照组用户在abtest上线前后点击率
评估实验组用户在新旧两种策略推荐图片的点击率

测试一定时间后，交换对照组和实验组用户，按照上述2个维度重新进行评估

参考文献

【1】Y Koren，R Bell，C Volinsky, “Matrix Factorization Techniques for Recommender Systems”, 《Computer》, 2009.08; 42(8):30-37

【2】洪亮劼, “知人知面需知心——人工智能技术在推荐系统中的应用”, 2016.11, http://mp.weixin.qq.com/s/JuaM8d52-f8AzTjEPnCl7g

【3】S. Funk, “Netflix Update: Try This at Home”, 2006.12, http://sifter.org/~simon/journal/20061211.html

【4】Y. Koren, “Factorization Meets the Neighborhood: A Mul-tifaceted Collaborative Filtering Model”, Proc. 14th ACM SIGKDD Int’l Conf. Knowledge Discovery and Data Mining, ACM Press, 2008, pp.426-434

【5】A. Paterek, “Improving Regularized Singular Value De-composition for Collaborative Filtering” Proc. KDD Cup and Workshop, ACM Press, 2007, pp.39-42

【6】G. Takács et al., “Major Components of the Gravity Recom- mendation System”, SIGKDD Explorations, 2007.09, vol.9, pp.80-84

【7】孟祥瑞, “ALS 在 Spark MLlib 中的实现”, 2015.05, http://www.csdn.net/article/2015-05-07/2824641

【8】Zhen-ming Yuan, et al., “A microblog recommendation algorithm based on social tagging and a temporal interest evolution model”, Frontiers of Information Technology & Electronic Engineering, 2015.07,
Volume 16, Issue 7, pp 532–540

【9】Z Zhao, Z Cheng, L Hong, EH Chi, “Improving User Topic Interest Profiles by Behavior Factorization”, Proceedings of the 24th International Conference on World Wide Web, 2015.05, pp.1406-1416

【10】阿里技术，”淘宝搜索/推荐系统背后深度强化学习与自适应在线学习的实践之路”, 2017.02, http://url.cn/451740J

【11】HT Cheng, L Koc, J Harmsen, T Shaked, “Wide & Deep Learning for Recommender Systems”, Proceedings of the 1st Workshop on Deep Learning for Recommender Systems, 2016.09, pp.7-10

【12】黄安埠, “递归的艺术 - 深度递归网络在序列式推荐的应用”, 2016.10, http://mp.weixin.qq.com/s?__biz=MzA3MDQ4MzQzMg==&mid=2665690422&idx=1&sn=9bd671983a85286149b51c908b686899&chksm=842bb9b1b35c30a7eedb8d03e173aa8f43465db90e11075ac0c73b1784582f21eb93dcbd3e65&scene=0%23wechat_redirect