1、通过ALS模型实现用户/商品Embedding的效果，获得其向量表示
准备训练数据， M = (U , I, R) 即 用户集U、商品集I、及评分数据R。
（1）商品集I的选择：可以根据业务目标确定商品候选集，比如TopK热度召回、或者流行度不高但在业务用户中区分度比较高的商品集等。个人建议量级控制在5W内，1W-2W左右比较合适，太大的话，用户产生行为的商品比较少，评分数据会非常的稀疏。
（2）用户集U的选择： 最好是粗召回策略确定的用户范围，因为ALS模型会生成所有U用户的特征向量表示，对于没有见过的用户u，没有其向量表示，其推荐也是冷启动策略。这里可以根据业务需要限制一个大范围，比如4000W-5000W的或大几百万的用户（从计算效率和内存使用上，个人建议500W内比较合适）。比如用户U定义为某些类目下购买人群、或者近期活跃人群等符合业务人群目标的潜在客户群。模型训练完之后，也是在这个用户集U中筛选出TopK相似的用户做推荐或扩量。
（3）评分数据R的选择：我们能采集到的大多是隐式反馈的数据，比如购买行为、浏览行为、收藏行为等。确定了U、I，确定了评分指标类型，就可以统计一段时间内，U对I的反馈数据R。数据量级大约在7亿条-10亿条，在模型参数设置合理的情况下，大约20-30分钟就可以训练完。
from pyspark.ml.recommendation import ALS
from pyspark.sql.functions import expr, isnull
""" ALS模型参数解读，和大小设置建议：
:param
rank=10, maxIter=10, regParam=0.1, numUserBlocks=10,
numItemBlocks=10, implicitPrefs=False, alpha=1.0, userCol="user", itemCol="item",
seed=None, ratingCol="rating", nonnegative=False, checkpointInterval=10,
intermediateStorageLevel="MEMORY_AND_DISK",
finalStorageLevel="MEMORY_AND_DISK", coldStartStrategy="nan", blockSize=4096
NumBlocks分块数：分块是为了并行计算，默认为10。可以根据数据量级适当放大，比如20。 可以对 numUserBlocks\numItemBlocks 单独进行配置并行度 ，也可以通过setNumBlocks(30)一起设置。
正则化参数：默认为1。
秩rank：模型中隐藏因子的个数，默认是10。即特征向量的维度。
implicitPrefs：显式偏好信息-false，隐式偏好信息-true，默认false(显示) 。 电商场景中 购买、点击、分享，都是隐式反馈。
alpha：隐式反馈时的置信度参数，默认是1.0。只用于隐式的偏好数据。
setMaxIter(10)：最大迭代次数,设置太大发生java.lang.StackOverflowError。建议范围 10 ～20。 超过20，比较容易失败。
coldStartStrategy: 预测时冷启动策略。默认是nan, 可以选择 drop。
"""
ratings = spark.sql("""
select
user_acct, user_id, main_sku_id, item_id, rating
from dmb_dev.dmb_dev_als_model_rating_matrix
""").repartition(3600)
train_data, test_data = ratings.randomSplit([0.9, 0.1], seed=4226)
train_data.cache()
als = ALS() \
.setImplicitPrefs(True) \
.setAlpha(0.7) \
.setMaxIter(20) \
.setRank(10) \
.setRegParam(0.01) \
.setNumBlocks(30) \
.setUserCol("user_id") \
.setItemCol("item_id") \
.setRatingCol("rating") \
.setColdStartStrategy("drop")
print(als.explainParams())
als_model = als.fit(train_data)
als_model.write().overwrite().save(model_save_path)
# 训练集合所有用户U的向量表示
candidate_user_factors = als_model.userFactors.withColumnRenamed("id", "user_id")\
.join(train_data.select("user_acct", "user_id").dropDuplicates(), ["user_id"])\
.withColumn("bin_group", expr("round(rand(),1)"))
candidate_user_factors.cache()
candidate_user_factors.write.format("orc").mode("overwrite")\
.saveAsTable("dev.dev_als_model_all_trained_users_factor_result")
train_data.unpersist()
# query用户的向量表示
target_user_factors = spark.sql("""
select
user_acct, user_id
from dev.dev_wdy_als_seed_users_table
group by user_acct, user_id
""").join(candidate_user_factors, ["user_acct", "user_id"])
target_user_factors.cache()
target_user_factors.write.format("orc").mode("overwrite")\
.saveAsTable("dev.dev_als_model_seed_users_factor")
# 候选用户向量表示
search_user_factors = candidate_user_factors.join(target_user_factors,
candidate_user_factors["user_acct"] == target_user_factors["user_acct"],
"left_outer")\
.where(isnull(target_user_factors["user_acct"]))\
.select(candidate_user_factors["user_acct"], candidate_user_factors["user_id"],
candidate_user_factors["features"], candidate_user_factors["bin_group"])
search_user_factors.write.format("orc").mode("overwrite")\
.saveAsTable("dev.dev_als_model_candidate_users_factor")
candidate_user_factors.unpersist()
target_user_factors.unpersist()
2、通过Faiss快速实现向量TopK相似检索
如果没有装faiss，可以选择安装CPU/GPU版本， pip install faiss-cpu
关于faiss的使用说明，可以参考向量数据库入坑指南：聊聊来自元宇宙大厂 Meta 的相似度检索技术 Faiss - 知乎
faiss来自facebook 开源 Meta Research · GitHub的github库为：GitHub - facebookresearch/faiss: A library for efficient similarity search and clustering of dense vectors.
根据业务需求的查询速度、精准度要求来选择合适的Faiss TopK向量查询方法。
# 判断 npy文件是否存在，不存在则执行以下操作；否则跳过此步骤，直接读取文件。
user_embedding = spark.sql("""
select
features[0],features[1],features[2],features[3],features[4],
features[5],features[6],features[7],features[8],features[9]
from
dev.dev_als_model_candidate_users_factor
where bin_group=0.1""").toPandas()
# 量级500W内执行顺利，再大的量级容易内存溢出失败。
np.save("user_embedding_01.npy", np.array(user_embedding, order='C'))
user_embedding = np.load("user_embedding_01.npy")
print("user_embedding data sample:", user_embedding[:3])
print("user embedding shape", user_embedding.shape)
dimension = user_embedding.shape[1]
nums_user = user_embedding.shape[0]
faiss.normalize_L2(user_embedding)
index = faiss.IndexFlatIP(dimension)
index.add(user_embedding)
print("index is trained:", index.is_trained)
print("index n total:", index.ntotal)
# 判断文件是否存在，如果存在则直接读取，否则先下载保存到本地。
## 这里k=30 或更大时，查询易失败。 k=20， 查询耗时久，但会成功，大约3小时。 k=10时，
k = 5
query1 = spark.sql("""select features[0],features[1],features[2],features[3],features[4],features[5],features[6],features[7],
features[8],features[9] from dev.dev_als_model_seed_users_factor""").toPandas()
np.save("query.npy", np.array(query1, order='C'))
query = np.load("query.npy")
print("query shape:", query.shape)
# 查询
t0 = time.time()
Deg, Ind = index.search(query, k)
t1 = time.time()
print("平均耗时 %7.3f min" % ((t1 - t0)/60))
# 保存索引
faiss.write_index(index, "faiss_01.index")
np.save("Ind_01.npy", Ind)
np.save("Deg_01.npy", Deg)
res = []
for i in range(query.shape[0]):
q_vector = query[i]
r_list = Ind[i]
for j in range(len(r_list)):
r_vector = user_embedding[r_list][j]
sim = Deg[i][j]
res.append(([float(v) for v in r_vector], float(sim)))
res = spark.createDataFrame(res, ["recommend_vector", "similarity"]).repartition(10)
res.cache()
res.write.format("orc").mode("overwrite")\
.saveAsTable("dev.dev_als_model_recommend_vector_result")
user_embedding = spark.sql("""
select
*
from
dev.dev_als_model_candidate_users_factor
where bin_group=0.1""")
res.join(user_embedding, res["recommend_vector"] == user_embedding["features"])\
.write.format("orc").mode("overwrite")\
.saveAsTable("dev.dev_als_model_recommend_user_pin_result")
查询速度实验对比数据：
| IndexIVFFlat |
IndexFlatIP |
IndexFlatIP |
| user embedding shape (4474857, 10) |
user embedding shape (4474857, 10) |
user embedding shape (4474857, 10) |
| query shape: (78525, 10) |
query shape: (78525, 10) |
query shape: (34525, 10) |
| k=5 |
k=5 |
k=10 |
| 平均耗时 10.522 min |
平均耗时 > 6h |
平均耗时 3h-4h |
业务中查询的候选集可能有4000W-5000W，而且对于查询响应时间有要求，使用IndexIVFFlat更符合上线需求。
nlist = 50
quantizer = faiss.IndexFlatL2(dimension)
index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_L2)
assert not index.is_trained
index.train(user_embedding)
assert index.is_trained
index.add(user_embedding) # 添加索引可能会有一点慢
index.nprobe = 10 # 默认 nprobe 是1 ,这里设置为10
3、通过I2I2U 或者 I2U2U来获得用户扩量结果
上述实现的是U2U的扩量方法，使用的是User-Factor向量表示。第一个U来自于业务营销目标I下的历史已购人群。即这是一个I2U2U的扩量方法。 I (目标商品)---> U(历史购买) ---> U(TopK相似) 。
当然也可以通过使用Item-Factor向量表示，实现 I2I2U，即 I (目标商品)---> I(TopK相似) ---> U(历史购买) ，这样来做商品相似召回，实现用户的扩量。
基于实验效果，或历史数据的验证来选择使用哪种方法投产。
4、算法设计框架总结
可以看到，这个算法设计框架其实是 Embedding + Faiss ，即用户/商品的向量表示 + Faiss快速向量相似检索 的设计模式。
那么第一部分的ALS模型当然可以替换成任何一种可以效果更好的Embedding算法模型，比如BERT 、Transformer等深度学习模型。而第二部分Faiss的查询可以保持不动，只要替换查询数据源就可以了。当然也可以将其优化成GPU的，或更快速的查询方式，以满足线上业务的需求。
但整体的算法设计框架是不变的，Embedding向量化 + Faiss相似检索。
Done.
