如何在Python中使用Elasticsearch和MongoDB实现个性化推荐

from elasticsearch import Elasticsearch
es = Elasticsearch()
# 索引名称
index_name = "users"
# 删除索引
if es.indices.exists(index=index_name):
    es.indices.delete(index=index_name)
# 创建索引
es.indices.create(index=index_name)
# 将用户行为表存入Elasticsearch
for row in behavior_rows:
    id = f"{row['user_id']}_{row['item_id']}"
    doc = {
        "user_id": row["user_id"],
        "item_id": row["item_id"],
        "behavior_type": row["behavior_type"]
    }
    es.index(index=index_name, id=id, body=doc)
# 将商品信息表存入Elasticsearch
for row in item_rows:
    id = row["item_id"]
    doc = {
        "item_name": row["item_name"]
    }
    es.index(index=index_name, id=id, body=doc)

# 聚合函数，计算每个用户对每个商品的行为得分
def aggregation_function(query):
    return {
        "score": {
            "scripted_metric": {
                "init_script": "state.transactions = []",
                "map_script": "state.transactions.add(doc['behavior_type'].value)",
                "combine_script": "float positive = 0; float negative = 0; for (t in state.transactions) {if (t == 'browse') {positive += 1} else {negative += 1}} return (positive + 1)/(positive + negative + 2)",
                "reduce_script": "float positive = 0; float negative = 0; for (s in states) {positive += s.transactions[0]; negative += s.transactions[1];} return (positive + 1)/(positive + negative + 2);"
            }
        }
    }
# 查询函数，查询每个用户对每个商品的行为得分
def query_function(user_id):
    query = {"query": {"term": {"user_id": user_id}}, "size": 0, "aggs": {"items": {"terms": {"field": "item_id.keyword"}, "aggs": aggregation_function(query)}}}
    return es.search(index=index_name, body=query)
# 计算每个用户对每个商品的行为得分，并存入Elasticsearch
for user_id in users:
    result = query_function(user_id)
    for item in result["aggregations"]["items"]["buckets"]:
        id = f"{user_id}_{item['key']}"
        doc = {
            "user_id": user_id,
            "item_id": item["key"],
            "score": item["score"]["value"]
        }
        es.index(index=index_name, id=id, body=doc)

# 查询函数，查询每个用户的推荐商品
def recommend_function(user_id):
    query = {"query": {"term": {"user_id": user_id}}, "sort": [{"score": {"order": "desc"}}], "size": 10, "_source": ["item_id"]}
    result = es.search(index=index_name, body=query)
    return [hit["_source"]["item_id"] for hit in result["hits"]["hits"]]
# 查询每个用户的推荐商品
for user_id in users:
    items = recommend_function(user_id)
    print(f"用户{user_id}的推荐商品是{items}")

from pymongo import MongoClient
client = MongoClient()
# 数据库名称和集合名称
db_name = "users"
collection_name = "behaviors"
# 删除集合
client[db_name][collection_name].drop()
# 将用户行为表存入MongoDB
client[db_name][collection_name].insert_many(behavior_rows)
# 将商品信息表存入MongoDB
client[db_name]["items"].insert_many(item_rows)

# 聚合函数，计算每个用户对每个商品的行为得分
def aggregation_function():
    return {"$add": [{"$divide": [{"$add": [{"$sum": [{"$cond": [{"$eq": ["$behavior_type", "browse"]}, 1, 0]}]}, 1]}, {"$add": [{"$sum": [{"$cond": [{"$eq": ["$behavior_type", "browse"]}, 1, 0]}]}, {"$sum": [{"$cond": [{"$eq": ["$behavior_type", "buy"]}, 1, 0]}]}]}]}, {"$literal": 0.00001}]}
# 查询函数，查询每个用户对每个商品的行为得分
def query_function(user_id):
    query = [{"$match": {"user_id": user_id}}, {"$group": {"_id": {"item_id": "$item_id", "user_id": "$user_id"}, "score": aggregation_function()}}, {"$out": f"tmp_{user_id}"}]
    client[db_name][collection_name].aggregate(query).allowDiskUse(True)
# 计算每个用户对每个商品的行为得分，并存入MongoDB
for user_id in users:
    query_function(user_id)
    client[db_name]["items"].create_index([("item_id", 1)])
    client[db_name][f"tmp_{user_id}"].create_index([("score", -1)])
    client[db_name][f"tmp_{user_id}"].aggregate([{"$sort": {"score": -1}}, {"$limit": 10}, {"$project": {"_id": 0, "item_id": "$_id.item_id"}}])
    client[db_name][f"tmp_{user_id}"].drop()

# 查询函数，查询每个用户的推荐商品
def recommend_function(user_id):
    query = {"_id.user_id": user_id}
    result = client[db_name]["tmp"].find(query, {"_id.item_id": 1}).sort("score", -1).limit(10)
    return [doc["_id"]["item_id"] for doc in result]
# 查询每个用户的推荐商品
for user_id in users:
    items = recommend_function(user_id)
    print(f"用户{user_id}的推荐商品是{items}")