Jul
24
2018
Customizing PredictionIO Similar Product Template to recommend users
PredictionIO is an open source Machine Learning Server provide by Apache. It is packed with various templates that can be used to quickly deploy a full scale recommendation engine.
The templates provided by PredictionIO can be easily customized depending upon your requirement.
In this blog we will modify the Similar Product Template, which used to recommend similar items to a user to recommend similar user to a user, i.e we will change user-to-item recommendation to user to user recommendation.
Note : This blog assumes that you have predictionIO event server up and running.
Step 1 :Create a new Engine from the Similar Product Engine Template.
<pre>
$ git clone https://github.com/apache/predictionio-template-similar-product.git userRecommendation
$ cd userRecommendation
</pre>
Step 2 : Open Engine.scala and modify the following lines:
We will change the Query class to get rid of items and put users in place of it. We will remove categories.
<pre>
case class Query(
users: List[String],
num: Int,
whiteList: Option[Set[String]],
blackList: Option[Set[String]]
)
</pre>
Get rid of ItemScore class and create a new class and name it to SimilarUserScore.
<pre>
case class SimilarUserScore(
user: String,
score: Double
)
</pre>
Step 3 :Open DataSource.scala and modify the following line :
We will change viewEvent case class to followEvent and remove the Item case class..
<pre>
case class FollowEvent(user: String, followedUser: String, t: Long)
</pre>
Now we will modify TrainingData class to use the newly created followEvent class created in above.
<pre>
class TrainingData(
val users: RDD[(String, User)],
val followEvents: RDD[FollowEvent] // MODIFIED
) extends Serializable {
override def toString = {
s”users: [${users.count()} (${users.take(2).toList}…)]” +
// MODIFIED
s”followEvents: [${followEvents.count()}] (${followEvents.take(2).toList}…)”
}
}
</pre>
The next modification in this file is to modify readTraining() function to read “follow events” and remove RDD of (entityId,Item).
<pre>
override
def readTraining(sc: SparkContext): TrainingData = {
// create a RDD of (entityID, User)
val usersRDD: RDD[(String, User)] = …
// MODIFIED
// get all “user” “follow” “followedUser” events
val followEventsRDD: RDD[FollowEvent] = PEventStore.find(
appName = dsp.appName,
entityType = Some(“user”),
eventNames = Some(List(“follow”)),
// targetEntityType is optional field of an event.
targetEntityType = Some(Some(“user”)))(sc)
// eventsDb.find() returns RDD[Event]
.map { event =>
val followEvent = try {
event.event match {
case “follow” => FollowEvent(
user = event.entityId,
followedUser = event.targetEntityId.get,
t = event.eventTime.getMillis)
case _ => throw new Exception(s”Unexpected event $event is read.”)
}
} catch {
case e: Exception => {
logger.error(s”Cannot convert $event to FollowEvent.” +
s” Exception: $e.”)
throw e
}
}
followEvent
}.cache()
new TrainingData(
users = usersRDD,
followEvents = followEventsRDD // MODIFIED
)
}
</pre>
Step 4 : Open Preparator.scala and modify the following lines :
This class is responsible for passing on the data to to the algorithm responsible for finding the similar users.
We will modify this class to pass the follow events as prepared data to the algorithm.
Modify the Preparator’s parpare() method:
<pre>
def prepare(sc: SparkContext, trainingData: TrainingData): PreparedData = {
new PreparedData(
users = trainingData.users,
followEvents = trainingData.followEvents) // MODIFIED
}
</pre>
Now we will modify the PrepareData class :
<pre>
class PreparedData(
val users: RDD[(String, User)],
val followEvents: RDD[FollowEvent] // MODIFIED
) extends Serializable
</pre>
Step 5: Modify ALSAlgorithm.scala :
This class is the core of our recommendation engine. We will modify ALSModel class to use similar user. We need to modify two events namely train() and predict().
Modify train() method to train with the follow event.
<pre>
def train(sc: SparkContext, data: PreparedData): ALSModel = {
require(data.followEvents.take(1).nonEmpty,
s”followEvents in PreparedData cannot be empty.” +
” Please check if DataSource generates TrainingData” +
” and Preprator generates PreparedData correctly.”)
require(data.users.take(1).nonEmpty,
s”users in PreparedData cannot be empty.” +
” Please check if DataSource generates TrainingData” +
” and Preprator generates PreparedData correctly.”)
// create User String ID to integer index BiMap
val userStringIntMap = BiMap.stringInt(data.users.keys)
val similarUserStringIntMap = userStringIntMap
// collect SimilarUser as Map and convert ID to Int index
val similarUsers: Map[Int, User] = data.users.map { case (id, similarUser) =>
(similarUserStringIntMap(id), similarUser)
}.collectAsMap().toMap
val mllibRatings = data.followEvents
.map { r =>
// Convert user and user String IDs to Int index for MLlib
val uindex = userStringIntMap.getOrElse(r.user, -1)
val iindex = similarUserStringIntMap.getOrElse(r.followedUser, -1)
if (uindex == -1)
logger.info(s”Couldn’t convert nonexistent user ID ${r.user}”
+ ” to Int index.”)
if (iindex == -1)
logger.info(s”Couldn’t convert nonexistent followedUser ID ${r.followedUser}”
+ ” to Int index.”)
((uindex, iindex), 1)
}.filter { case ((u, i), v) =>
// keep events with valid user and user index
(u != -1) && (i != -1)
}
.map { case ((u, i), v) =>
// MLlibRating requires integer index for user and user
MLlibRating(u, i, v)
}
.cache()
// MLLib ALS cannot handle empty training data.
require(mllibRatings.take(1).nonEmpty,
s”mllibRatings cannot be empty.” +
” Please check if your events contain valid user and followedUser ID.”)
// seed for MLlib ALS
val seed = ap.seed.getOrElse(System.nanoTime)
val m = ALS.trainImplicit(
ratings = mllibRatings,
rank = ap.rank,
iterations = ap.numIterations,
lambda = ap.lambda,
blocks = -1,
alpha = 1.0,
seed = seed)
new ALSModel(
similarUserFeatures = m.productFeatures.collectAsMap().toMap,
similarUserStringIntMap = similarUserStringIntMap,
similarUsers = similarUsers
)
}
</pre>
Modify predict() method to predict similar users
<pre>
def predict(model: ALSModel, query: Query): PredictedResult = {
val similarUserFeatures = model.similarUserFeatures
// convert similarUsers to Int index
val queryList: Set[Int] = query.users.map(model.similarUserStringIntMap.get)
.flatten.toSet
val queryFeatures: Vector[Array[Double]] = queryList.toVector
// similarUserFeatures may not contain the requested user
.map { similarUser => similarUserFeatures.get(similarUser) }
.flatten
val whiteList: Option[Set[Int]] = query.whiteList.map( set =>
set.map(model.similarUserStringIntMap.get).flatten
)
val blackList: Option[Set[Int]] = query.blackList.map ( set =>
set.map(model.similarUserStringIntMap.get).flatten
)
val ord = Ordering.by[(Int, Double), Double](_._2).reverse
val indexScores: Array[(Int, Double)] = if (queryFeatures.isEmpty) {
logger.info(s”No similarUserFeatures vector for query users ${query.users}.”)
Array[(Int, Double)]()
} else {
similarUserFeatures.par // convert to parallel collection
.mapValues { f =>
queryFeatures.map { qf =>
cosine(qf, f)
}.sum
}
.filter(_._2 > 0) // keep similarUsers with score > 0
.seq // convert back to sequential collection
.toArray
}
val filteredScore = indexScores.view.filter { case (i, v) =>
isCandidateSimilarUser(
i = i,
similarUsers = model.similarUsers,
queryList = queryList,
whiteList = whiteList,
blackList = blackList
)
}
val topScores = getTopN(filteredScore, query.num)(ord).toArray
val similarUserScores = topScores.map { case (i, s) =>
SimilarUserScore(
user = model.similarUserIntStringMap(i),
score = s
)
}
PredictedResult(similarUserScores)
}
</pre>.
Step 6 : Test results :
- Build the source code by running :
$pio build - Train the engine by running
$pio train - Deploy the engine by running
$pio deploy
Now we can query the engine by sending a GET/POST request :
<pre>
$ curl -H “Content-Type: application/json” \
-d ‘{ “users”: [“u1”], “num”: 4 }’ \
http://localhost:8000/queries.json
</pre>
Sample output :
<pre>
{
“similarUserScores”:[
{“user”:”u3″,”score”:0.7574200014043541},
{“user”:”u10″,”score”:0.6484507108863744},
{“user”:”u43″,”score”:0.64741489488357},
{“user”:”u29″,”score”:0.5767264820728124}
]
}
</pre>
So with these simple customizations we are able to change the nature of similar product template to predict users in place of items.
