[Spark机器学习]基于Spark 2.0 机器学习之推荐系统实现

1 什么是ALS

ALS

alternating least squares

ALS

8*8

（u1,u2,…,u8）

（v1,v2,…,v8）

1-9

（u6，v5）

1

（u6，v5）

ALS

m*n

U（m*k）

V（k*n）

ALS

O(mn)

O((m+n)k)

ALS

ui

vj

ALS

A

spark

ALS

2 spark中ALS的实现原理

Spark

2.1 介绍

Spark

Netflix

preference

confidence

2.2 显式反馈模型

ratings

SVD

u

ui

v

vj

rating

lambda

2.3 隐式反馈模型

u

v

beliefs

confidence

ui

vj

u，v

u，v

factors

2.4 求解最小化损失函数

m*n

m

n

m*n

3 ALS在spark中的实现

spark

ALS

org.apache.spark.ml.recommendation.ALS.scala

ALS

//处理训练数据
val data = sc.textFile("data/mllib/als/test.data")
val ratings = data.map(_.split(',') match { case Array(user, item, rate) =>
Rating(user.toInt, item.toInt, rate.toDouble)
})
// 使用ALS训练推荐模型
val rank = 10
val numIterations = 10
val model = ALS.train(ratings, rank, numIterations, 0.01)

ALS.scala

train

train

train

def train(
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): MatrixFactorizationModel

ratings

id

id

rank

numUserBlocks

numItemBlocks

maxIter

regParam

lambda

implicitPrefs

Nonnegative

Nonnegative

spark

train

ALSPartitioner

LocalIndexEncoder

ALSPartitioner

hash

id

hash

LocalIndexEncoder

（blockid，localindex）

（分区id，分区内索引）

ID

blockid

localindex

val userPart = new ALSPartitioner(numUserBlocks)
val itemPart = new ALSPartitioner(numItemBlocks)
val userLocalIndexEncoder = new LocalIndexEncoder(userPart.numPartitions)
val itemLocalIndexEncoder = new LocalIndexEncoder(itemPart.numPartitions)

//ALSPartitioner即HashPartitioner
class HashPartitioner(partitions: Int) extends Partitioner {
def numPartitions: Int = partitions
def getPartition(key: Any): Int = key match {
case null => 0
case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
}
override def equals(other: Any): Boolean = other match {
case h: HashPartitioner =>
h.numPartitions == numPartitions
case _ =>
false
}
override def hashCode: Int = numPartitions
}

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

private[this] final val numLocalIndexBits =
math.min(java.lang.Integer.numberOfLeadingZeros(numBlocks - 1), 31)
//左移（<<,相当于乘2），右移（>>，相当于除2）和无符号右移（>>>，无符号右移，忽略符号位，空位都以0补齐）
private[this] final val localIndexMask = (1 << numLocalIndexBits) - 1
//encodeIndex高位存分区ID，在低位存对应分区的索引
def encode(blockId: Int, localIndex: Int): Int = {
(blockId << numLocalIndexBits) | localIndex
}
@inline
def blockId(encoded: Int): Int = {
encoded >>> numLocalIndexBits
}

@inline
def localIndex(encoded: Int): Int = {
encoded & localIndexMask
}
}

nonnegative

nonnegative

true

NNLS

Cholesky

val solver = if (nonnegative) new NNLSSolver else new CholeskySolver

U

ml

dppsv

lapack.dppsv(“u”, k, 1, ne.ata, ne.atb, k, info)

dppsv

Fortran

`NNLS

ratings

ratings

（（用户分区id，商品分区id），分区数据集blocks））

ALSPartitioner

getPartitions

RatingBlock

（用户分区id，商品分区id ）

（用户id集，商品id集，打分集）

val blockRatings = partitionRatings(ratings, userPart, itemPart)
.persist(intermediateRDDStorageLevel)

//以下是partitionRatings的实现
//默认是10*10
val numPartitions = srcPart.numPartitions * dstPart.numPartitions
ratings.mapPartitions { iter =>
val builders = Array.fill(numPartitions)(new RatingBlockBuilder[ID])
iter.flatMap { r =>
val srcBlockId = srcPart.getPartition(r.user)
val dstBlockId = dstPart.getPartition(r.item)
//当前builder的索引位置
val idx = srcBlockId + srcPart.numPartitions * dstBlockId
val builder = builders(idx)
builder.add(r)
//如果某个builder的数量大于2048，那么构建一个分区
if (builder.size >= 2048) { // 2048 * (3 * 4) = 24k
builders(idx) = new RatingBlockBuilder
//单元素集合
Iterator.single(((srcBlockId, dstBlockId), builder.build()))
} else {
Iterator.empty
}
} ++ {
builders.view.zipWithIndex.filter(_._1.size > 0).map { case (block, idx) =>
//用户分区id
val srcBlockId = idx % srcPart.numPartitions
//商品分区id
val dstBlockId = idx / srcPart.numPartitions
((srcBlockId, dstBlockId), block.build())
}
}
}.groupByKey().mapValues { blocks =>
val builder = new RatingBlockBuilder[ID]
blocks.foreach(builder.merge)
builder.build()
}.setName("ratingBlocks")
}

inblocks

outblocks

inblocks

outblocks

v1

u1

u2

v2

u1

u2

u3

O(m*rank)

m

rank

v1

v2

u1

u2

u2

v1

v2

U

V

Q1

v1

InBlock

P1

u1

u2

Q1

u1

u1

spark

RDD

OutBlock

U

V

OutBlock

InBlock

V

U

OutBlock

InBlock

InBlock

OutBlock

RDD

InBlock

InBlock

InBlock

OutBlock

InBlock

OutBlock

val (userInBlocks, userOutBlocks) =makeBlocks("user", blockRatings,
userPart, itemPart,intermediateRDDStorageLevel)
//交换userBlockId和itemBlockId以及其对应的数据
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)

InBlock

OutBlock

makeBlocks

InBlock

val inBlocks = ratingBlocks.map {
case ((srcBlockId, dstBlockId), RatingBlock(srcIds, dstIds, ratings)) =>
val start = System.nanoTime()
val dstIdSet = new OpenHashSet[ID](1 << 20)
//将用户id保存到hashset中，用来去重
dstIds.foreach(dstIdSet.add)
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
}
//对用户id进行排序
Sorting.quickSort(sortedDstIds)
val dstIdToLocalIndex = new OpenHashMap[ID, Int](sortedDstIds.length)
i = 0
while (i < sortedDstIds.length) {
dstIdToLocalIndex.update(sortedDstIds(i), i)
i += 1
}
//求取块内，用户id的本地位置
val dstLocalIndices = dstIds.map(dstIdToLocalIndex.apply)
//返回数据集
(srcBlockId, (dstBlockId, srcIds, dstLocalIndices, ratings))
}.groupByKey(new ALSPartitioner(srcPart.numPartitions))
.mapValues { iter =>
val builder =
new UncompressedInBlockBuilder[ID](new LocalIndexEncoder(dstPart.numPartitions))
iter.foreach { case (dstBlockId, srcIds, dstLocalIndices, ratings) =>
builder.add(dstBlockId, srcIds, dstLocalIndices, ratings)
}
//构建非压缩块，并压缩为InBlock
builder.build().compress()
}.setName(prefix + "InBlocks")
.persist(storageLevel)

ratingBlocks

map

ratingBlocks

（商品分区id，（用户分区id，商品集合，用户id在分区中相对应的位置，打分）

groupByKey

id

key

key

（商品分区id，InBlocks）

InBlock

Q1

v1

v2

(v1, u1, r11)，(v2, u1, r12)， (v1, u2, r21)， (v2, u2, r22)， (v2, u3, r32)

Tuple

Tuple

Tuple

Tuple

Tuple

spark

([v1, v2, v1, v2, v2], [u1, u1, u2, u2, u3], [r11, r12, r21, r22, r32])

spark

LocalIndexEncoder

UncompressedInBlock

（商品id集，用户id集对应的编码集，打分集）

([v1, v2, v1, v2, v2], [ui1, ui1, ui2, ui2, ui3], [r11, r12, r21, r22, r32])

spark

compress

id

（不重复的有序的商品id集，商品位置偏移集，用户id集对应的编码集，打分集）

coo

csc

([v1, v2, v1, v2, v2], [ui1, ui1, ui2, ui2, ui3], [r11, r12, r21, r22, r32])

([v1, v2], [0, 2, 5], [ui1, ui2, ui1, ui2, ui3], [r11, r21, r12, r22, r32])

[0, 2]

v1

[0, 2]

[2, 5]

v2

[2, 5]

Compress

spark

Timsort

UncompressedInBlock

InBlock

def compress(): InBlock[ID] = {
val sz = length
//Timsort排序
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
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
i = 0
//计算偏移量
while (i < numUniqueSrdIds) {
sum += dstCounts(i)
i += 1
dstPtrs(i) = sum
}
InBlock(uniqueSrcIds, dstPtrs, dstEncodedIndices, ratings)
}
private def sort(): Unit = {
val sz = length
val sortId = Utils.random.nextInt()
val sorter = new Sorter(new UncompressedInBlockSort[ID])
sorter.sort(this, 0, length, Ordering[KeyWrapper[ID]])
}

OutBlock

val outBlocks = inBlocks.mapValues { case InBlock(srcIds, dstPtrs, dstEncodedIndices, _) =>
val encoder = new LocalIndexEncoder(dstPart.numPartitions)
val activeIds = Array.fill(dstPart.numPartitions)(mutable.ArrayBuilder.make[Int])
var i = 0
val seen = new Array[Boolean](dstPart.numPartitions)
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
seen(dstBlockId) = true
}
j += 1
}
i += 1
}
activeIds.map { x =>
x.result()
}
}.setName(prefix + "OutBlocks")
.persist(storageLevel)

inBlocks

（用户分区id，（不重复的用户id集，用户位置偏移集，商品id集对应的编码集，打分集））

activeIds

activeIds

OutBlocks

（用户分区id，OutBlocks）

OutBlock

InBlock

OutBlock

InBlock

var userFactors = initialize(userInBlocks, rank, seedGen.nextLong())
var itemFactors = initialize(itemInBlocks, rank, seedGen.nextLong())

userFactors

（用户分区id，用户特征矩阵factors）

factors

rank

itemFactors

computeFactors

inblock

outblock

outblock

userFactor

join

id

key

factor

（用户分区id集，用户分区对应的factor集）

inblock

merged

join

（商品inblock，（用户分区id集，用户分区对应的factor集））

val srcOut = srcOutBlocks.join(srcFactorBlocks).flatMap {
case (srcBlockId, (srcOutBlock, srcFactors)) =>
srcOutBlock.view.zipWithIndex.map { case (activeIndices, dstBlockId) =>
(dstBlockId, (srcBlockId, activeIndices.map(idx => srcFactors(idx))))
}
}
val merged = srcOut.groupByKey(new ALSPartitioner(dstInBlocks.partitions.length))
dstInBlocks.join(merged)

InBlock

InBlock

spark

InBlock

InBlock

id

while (j < dstIds.length) {
ls.reset()
var i = srcPtrs(j)
var numExplicits = 0
while (i < srcPtrs(j + 1)) {
val encoded = srcEncodedIndices(i)
val blockId = srcEncoder.blockId(encoded)
val localIndex = srcEncoder.localIndex(encoded)
val srcFactor = sortedSrcFactors(blockId)(localIndex)
val rating = ratings(i)
ls.add(srcFactor, rating)
numExplicits += 1
i += 1
}
dstFactors(j) = solver.solve(ls, numExplicits * regParam)
j += 1
}

YtY

//在循环之外计算
val YtY = if (implicitPrefs) Some(computeYtY(srcFactorBlocks, rank)) else None

//在每个循环内
if (implicitPrefs) {
ls.merge(YtY.get)
}
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)
}
}