Skip to content

Commit d580344

Browse files
dciborowdciborow
authored
feat: add string to sar
1 parent f3953bc commit d580344

7 files changed

Lines changed: 531 additions & 266 deletions

File tree

‎core/src/main/python/synapse/ml/recommendation/SAR.py‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
# Copyright (C) Microsoft Corporation. All rights reserved.
2+
# Licensed under the MIT License. See LICENSE in the project root for information.
3+
4+
import sys
5+
6+
if sys.version >= "3":
7+
basestring = str
8+
9+
from synapse.ml.core.schema.Utils import *
10+
from synapse.ml.recommendation._SAR import _SAR
11+
12+
@inherit_doc
13+
class SAR(_SAR):
14+
def __init__(self, **kwargs):
15+
_SAR.__init__(self, **kwargs)
16+
17+
def calculateUserItemAffinities(self, dataset):
18+
if dataset.schema[self.getUserCol()].dataType == StringType():
19+
dataset = dataset.withColumn(self.getUserCol(), dataset[self.getUserCol()].cast("int"))
20+
if dataset.schema[self.getItemCol()].dataType == StringType():
21+
dataset = dataset.withColumn(self.getItemCol(), dataset[self.getItemCol()].cast("int"))
22+
return self._call_java("calculateUserItemAffinities", dataset)
23+
24+
def calculateItemItemSimilarity(self, dataset):
25+
if dataset.schema[self.getUserCol()].dataType == StringType():
26+
dataset = dataset.withColumn(self.getUserCol(), dataset[self.getUserCol()].cast("int"))
27+
if dataset.schema[self.getItemCol()].dataType == StringType():
28+
dataset = dataset.withColumn(self.getItemCol(), dataset[self.getItemCol()].cast("int"))
29+
return self._call_java("calculateItemItemSimilarity", dataset)

‎core/src/main/python/synapse/ml/recommendation/SARModel.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,8 @@
1515
class SARModel(_SARModel):
1616
def recommendForAllUsers(self, numItems):
1717
return self._call_java("recommendForAllUsers", numItems)
18+
19+
def recommendForUserSubset(self, dataset, numItems):
20+
if dataset.schema[self.getUserCol()].dataType == StringType():
21+
dataset = dataset.withColumn(self.getUserCol(), dataset[self.getUserCol()].cast("int"))
22+
return self._call_java("recommendForUserSubset", dataset, numItems)

‎core/src/main/scala/com/microsoft/azure/synapse/ml/recommendation/SAR.scala‎

Lines changed: 32 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,3 @@
1-
// Copyright (C) Microsoft Corporation. All rights reserved.
2-
// Licensed under the MIT License. See LICENSE in project root for information.
3-
41
package com.microsoft.azure.synapse.ml.recommendation
52

63
import breeze.linalg.{CSCMatrix => BSM}
@@ -13,7 +10,7 @@ import org.apache.spark.ml.util.{DefaultParamsReadable, DefaultParamsWritable, I
1310
import org.apache.spark.mllib.linalg
1411
import org.apache.spark.mllib.linalg.{DenseVector, Matrices, SparseMatrix}
1512
import org.apache.spark.sql.functions.{col, collect_list, sum, udf, _}
16-
import org.apache.spark.sql.types.StructType
13+
import org.apache.spark.sql.types.{StringType, StructType}
1714
import org.apache.spark.sql.{DataFrame, Dataset}
1815

1916
import java.text.SimpleDateFormat
@@ -106,8 +103,22 @@ class SAR(override val uid: String) extends Estimator[SARModel]
106103
(0 to numItems.value).map(i => map.getOrElse(i, 0.0).toFloat).toArray
107104
})
108105

109-
dataset
110-
.withColumn(C.AffinityCol, (dataset.columns.contains(getTimeCol), dataset.columns.contains(getRatingCol)) match {
106+
val userColType = dataset.schema(getUserCol).dataType
107+
val itemColType = dataset.schema(getItemCol).dataType
108+
109+
val castedDataset = (userColType, itemColType) match {
110+
case (StringType, StringType) =>
111+
dataset.withColumn(getUserCol, col(getUserCol).cast("int"))
112+
.withColumn(getItemCol, col(getItemCol).cast("int"))
113+
case (StringType, _) =>
114+
dataset.withColumn(getUserCol, col(getUserCol).cast("int"))
115+
case (_, StringType) =>
116+
dataset.withColumn(getItemCol, col(getItemCol).cast("int"))
117+
case _ => dataset
118+
}
119+
120+
castedDataset
121+
.withColumn(C.AffinityCol, (castedDataset.columns.contains(getTimeCol), castedDataset.columns.contains(getRatingCol)) match {
111122
case (true, true) => blendWeights(timeDecay(col(getTimeCol)), col(getRatingCol))
112123
case (true, false) => timeDecay(col(getTimeCol))
113124
case (false, true) => col(getRatingCol)
@@ -197,7 +208,21 @@ class SAR(override val uid: String) extends Estimator[SARModel]
197208
})
198209
})
199210

200-
dataset
211+
val userColType = dataset.schema(getUserCol).dataType
212+
val itemColType = dataset.schema(getItemCol).dataType
213+
214+
val castedDataset = (userColType, itemColType) match {
215+
case (StringType, StringType) =>
216+
dataset.withColumn(getUserCol, col(getUserCol).cast("int"))
217+
.withColumn(getItemCol, col(getItemCol).cast("int"))
218+
case (StringType, _) =>
219+
dataset.withColumn(getUserCol, col(getUserCol).cast("int"))
220+
case (_, StringType) =>
221+
dataset.withColumn(getItemCol, col(getItemCol).cast("int"))
222+
case _ => dataset
223+
}
224+
225+
castedDataset
201226
.select(col(getItemCol), col(getUserCol))
202227
.groupBy(getItemCol).agg(collect_list(getUserCol) as "collect_list")
203228
.withColumn(C.FeaturesCol, createItemFeaturesVector(col("collect_list")))

‎core/src/main/scala/com/microsoft/azure/synapse/ml/recommendation/SARModel.scala‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,3 @@
1-
// Copyright (C) Microsoft Corporation. All rights reserved.
2-
// Licensed under the MIT License. See LICENSE in project root for information.
3-
41
package com.microsoft.azure.synapse.ml.recommendation
52

63
import com.microsoft.azure.synapse.ml.codegen.Wrappable

‎core/src/test/python/synapsemltest/recommendation/test_ranking.py‎

Lines changed: 55 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -67,10 +67,52 @@
6767
.cache()
6868
)
6969

70+
ratings_with_strings = (
71+
spark.createDataFrame(
72+
[
73+
("user0", "item1", 4, 4),
74+
("user0", "item3", 1, 1),
75+
("user0", "item4", 5, 5),
76+
("user0", "item5", 3, 3),
77+
("user0", "item7", 3, 3),
78+
("user0", "item9", 3, 3),
79+
("user0", "item10", 3, 3),
80+
("user1", "item1", 4, 4),
81+
("user1", "item2", 5, 5),
82+
("user1", "item3", 1, 1),
83+
("user1", "item6", 4, 4),
84+
("user1", "item7", 5, 5),
85+
("user1", "item8", 1, 1),
86+
("user1", "item10", 3, 3),
87+
("user2", "item1", 4, 4),
88+
("user2", "item2", 1, 1),
89+
("user2", "item3", 1, 1),
90+
("user2", "item4", 5, 5),
91+
("user2", "item5", 3, 3),
92+
("user2", "item6", 4, 4),
93+
("user2", "item8", 1, 1),
94+
("user2", "item9", 5, 5),
95+
("user2", "item10", 3, 3),
96+
("user3", "item2", 5, 5),
97+
("user3", "item3", 1, 1),
98+
("user3", "item4", 5, 5),
99+
("user3", "item5", 3, 3),
100+
("user3", "item6", 4, 4),
101+
("user3", "item7", 5, 5),
102+
("user3", "item8", 1, 1),
103+
("user3", "item9", 5, 5),
104+
("user3", "item10", 3, 3),
105+
],
106+
["originalCustomerID", "newCategoryID", "rating", "notTime"],
107+
)
108+
.coalesce(1)
109+
.cache()
110+
)
111+
70112

71113
class RankingSpec(unittest.TestCase):
72114
@staticmethod
73-
def adapter_evaluator(algo):
115+
def adapter_evaluator(algo, data):
74116
recommendation_indexer = RecommendationIndexer(
75117
userInputCol=USER_ID,
76118
userOutputCol=USER_ID_INDEX,
@@ -80,7 +122,7 @@ def adapter_evaluator(algo):
80122

81123
adapter = RankingAdapter(mode="allUsers", k=5, recommender=algo)
82124
pipeline = Pipeline(stages=[recommendation_indexer, adapter])
83-
output = pipeline.fit(ratings).transform(ratings)
125+
output = pipeline.fit(data).transform(data)
84126
print(str(output.take(1)) + "\n")
85127

86128
metrics = ["ndcgAt", "fcp", "mrr"]
@@ -91,13 +133,17 @@ def adapter_evaluator(algo):
91133
+ str(RankingEvaluator(k=3, metricName=metric).evaluate(output)),
92134
)
93135

94-
# def test_adapter_evaluator_als(self):
95-
# als = ALS(userCol=USER_ID_INDEX, itemCol=ITEM_ID_INDEX, ratingCol=RATING_ID)
96-
# self.adapter_evaluator(als)
97-
#
98-
# def test_adapter_evaluator_sar(self):
99-
# sar = SAR(userCol=USER_ID_INDEX, itemCol=ITEM_ID_INDEX, ratingCol=RATING_ID)
100-
# self.adapter_evaluator(sar)
136+
def test_adapter_evaluator_als(self):
137+
als = ALS(userCol=USER_ID_INDEX, itemCol=ITEM_ID_INDEX, ratingCol=RATING_ID)
138+
self.adapter_evaluator(als, ratings)
139+
140+
def test_adapter_evaluator_sar(self):
141+
sar = SAR(userCol=USER_ID_INDEX, itemCol=ITEM_ID_INDEX, ratingCol=RATING_ID)
142+
self.adapter_evaluator(sar, ratings)
143+
144+
def test_adapter_evaluator_sar_with_strings(self):
145+
sar = SAR(userCol=USER_ID_INDEX, itemCol=ITEM_ID_INDEX, ratingCol=RATING_ID)
146+
self.adapter_evaluator(sar, ratings_with_strings)
101147

102148
def test_all_tiny(self):
103149
customer_index = StringIndexer(inputCol=USER_ID, outputCol=USER_ID_INDEX)

0 commit comments

Comments
 (0)