使用Spark ALS模型 + Faiss向量检索实现用户扩量实例

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模型参数解读，和大小设置建议：
:paramrank=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=4096NumBlocks分块数：分块是为了并行计算，默认为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("""selectuser_acct, user_id, main_sku_id, item_id, ratingfrom 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("""selectuser_acct, user_idfrom dev.dev_wdy_als_seed_users_tablegroup 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_factorwhere 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_factorwhere 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")