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

这篇具有很好参考价值的文章主要介绍了使用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模型参数解读,和大小设置建议:
: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.文章来源地址https://www.toymoban.com/news/detail-641087.html

到了这里,关于使用Spark ALS模型 + Faiss向量检索实现用户扩量实例的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【向量数据库】相似向量检索Faiss数据库的安装及余弦相似度计算(C++)

    Faiss 是一个强大的向量相似度搜索库,具有以下优点: 高效的搜索性能:Faiss 在处理大规模向量数据时表现出色。它利用了高度优化的索引结构和近似搜索算法,可以快速地执行最近邻搜索和相似度匹配,具有很低的查询延迟。 高度可扩展:Faiss 提供了多种索引结构和算法

    2024年02月07日
    浏览(55)
  • 向量检索(一)Faiss 在工业界的应用和常见问题解决

    传统的搜索,使用关键做精确的查找,利用倒排索引在索引库中搜索。日常在用的百度,Google都属于搜索。 在 AI 时代,我们需要查找一张相似的图片,一个问题的答案,或者根据一段音乐查找对应的歌曲,这些情况下没有准确的用来做检索。 这些图片,问题(

    2024年02月02日
    浏览(50)
  • 【深度学习】以图搜索- 2021sota repVgg来抽取向量 + facebook的faiss的做特征检索, 从环境搭建到运行案例从0到1

    Faiss的全称是Facebook AI Similarity Search。 这是一个开源库,针对高维空间中的海量数据,提供了高效且可靠的检索方法。 暴力检索耗时巨大,对于一个要求实时人脸识别的应用来说是不可取的。 而Faiss则为这种场景提供了一套解决方案。 Faiss从两个方面改善了暴力搜索算法存在

    2024年02月15日
    浏览(45)
  • spark基于HNSW向量检索

    参考文档:https://talks.anghami.com/blazing-fast-approximate-nearest-neighbour-search-on-apache-spark-using-hnsw/ HNSW参数调优文档:https://github.com/nmslib/hnswlib/blob/master/ALGO_PARAMS.md spark 运行HNSW向量检索分为以下三步 1 创建HNSW索引,并存储到磁盘 2 将存储的索引分发到每个executor 3 进行向量检索 使用

    2024年02月11日
    浏览(37)
  • 二 根据用户行为数据创建ALS模型并召回商品

    2.0 用户行为数据拆分 方便练习可以对数据做拆分处理 pandas的数据分批读取 chunk 厚厚的一块 相当大的数量或部分 2.1 预处理behavior_log数据集 创建spark session 从hdfs中加载csv文件为DataFrame 显示结果: 从hdfs加载数据为dataframe,并设置结构 显示结果: 分析数据集字段的类型和格式

    2024年02月12日
    浏览(34)
  • 向量数据库:usearch的简单使用+实现图片检索应用

    usearch是快速开源搜索和聚类引擎×,用于C++、C、Python、JavaScript、Rust、Java、Objective-C、Swift、C#、GoLang和Wolfram 🔍中的向量和🔜字符串× 一个简单的例子(注:本例子在运行时向index中不断添加项目,并将最后的index持久化为一个文件,在运行时由于添加项目内存占用会不断增

    2024年02月02日
    浏览(52)
  • Spring AI - 使用向量数据库实现检索式AI对话

     Spring AI 并不仅限于针对大语言模型对话API进行了统一封装,它还可以通过简单的方式实现LangChain的一些功能。本篇将带领读者实现一个简单的检索式AI对话接口。  在一些场景下,我们想让AI根据我们提供的数据进行回复。因为对话有最大Token的限制,因此很多场景下我们

    2024年04月14日
    浏览(52)
  • Spark MLlib ----- ALS算法

    在谈ALS(Alternating Least Squares)之前首先来谈谈LS,即最小二乘法。LS算法是ALS的基础,是一种数优化技术,也是一种常用的机器学习算法,他通过最小化误差平方和寻找数据的最佳匹配,利用最小二乘法寻找最优的未知数据,保证求的数据与已知的数据误差最小。LS也被用于拟

    2024年02月02日
    浏览(43)
  • Elasticsearch 8.X 向量检索和普通检索能否实现组合检索?如何实现?

    向量组合条件查询,报 [vector] malformed query, expected [END_OBJECT] but found [FIELD_NAME] 错误, 向量查询是不支持复合条件查询吗? ——问题来自:死磕 Elasticsearch 知识星球 https://t.zsxq.com/18skX0ZS6 类似问题在社群里被问到 2 次以上了! 向量搜索热度不减,所以我们非常有必要将向量搜

    2024年04月11日
    浏览(47)
  • 信息检索与数据挖掘 | (五)文档评分、词项权重计算及向量空间模型

    目录 📚词项频率及权重计算 🐇词项频率 🐇逆文档频率 🐇tf-idf权重计算 📚向量空间模型 🐇余弦相似度 🐇查询向量 🐇向量相似度计算 📚其他tf-idf权值计算方法 🐇tf的亚线性尺度变换方法 🐇基于最大值的tf归一化 🐇文档权值和查询权重机 我们需要一种方法分配一个

    2024年02月08日
    浏览(45)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包