推荐系统架构设计¶

from flask import Flask, request
import redis
import json

app = Flask(__name__)
redis_client = redis.Redis(host='localhost', port=6379)

@app.route('/recall', methods=['GET'])
def recall():
    """召回接口"""
    user_id = request.args.get('user_id')
    context = request.args.get('context')

    # 多路召回
    candidates = []

    # 1. CF召回
    cf_items = collaborative_filtering_recall(user_id)
    candidates.extend(cf_items)

    # 2. 内容召回
    content_items = content_based_recall(user_id)
    candidates.extend(content_items)

    # 3. 热门召回
    hot_items = hot_items_recall()
    candidates.extend(hot_items)

    # 4. 深度召回
    deep_items = deep_learning_recall(user_id)
    candidates.extend(deep_items)

    # 去重
    unique_candidates = list(set(candidates))

    # 限制数量
    final_candidates = unique_candidates[:500]  # 切片操作，取前n个元素

    return {'items': final_candidates}

@app.route('/rank', methods=['POST'])
def rank():
    """排序接口"""
    data = request.json
    user_id = data['user_id']
    items = data['items']
    context = data.get('context', {})

    # 1. 获取特征
    user_features = get_user_features(user_id)
    item_features_list = [get_item_features(item_id) for item_id in items]
    context_features = extract_context_features(context)

    # 2. 特征拼接
    features_list = []
    for item_features in item_features_list:
        features = {**user_features, **item_features, **context_features}
        features_list.append(features)

    # 3. 批量预测
    scores = model.predict_batch(features_list)

    # 4. 排序
    ranked_items = sorted(zip(items, scores), key=lambda x: x[1], reverse=True)  # zip按位置配对  # lambda匿名函数

    # 5. 返回Top N
    top_items = [item for item, score in ranked_items[:100]]

    return {'items': top_items}

# 离线特征计算
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("OfflineFeatureEngineering").getOrCreate()

# 读取数据
events = spark.read.parquet("/data/events/")

# 特征计算
user_features = events.groupBy("user_id").agg(
    count("event_id").alias("event_count"),
    avg("item_price").alias("avg_price")
)

# 保存特征
user_features.write.parquet("/data/features/user_features/")

# 实时特征更新
from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()

# 读取Kafka
events = env.add_source(kafka_source)

# 特征计算
features = events \
    .key_by(lambda x: x['user_id']) \
    .reduce(lambda a, b: merge_features(a, b))

# 写入Redis
features.add_sink(redis_sink)

env.execute("Real-time Feature Update")

# 召回服务
from flask import Flask

recall_app = Flask(__name__)

@recall_app.route('/recall')
def recall():
    return recall_items()

# 排序服务
rank_app = Flask(__name__)

@rank_app.route('/rank')
def rank():
    return rank_items()

# 特征服务
feature_app = Flask(__name__)

@feature_app.route('/features')
def features():
    return get_features()

from consul import Consul

consul = Consul(host='localhost', port=8500)

# 服务注册
consul.agent.service.register(
    name='recall-service',
    service_id='recall-service-1',
    address='localhost',
    port=8080,
    tags=['recall', 'v1']
)

# 服务发现
services = consul.agent.services()
recall_services = [s for s in services.values() if s['Service'] == 'recall-service']

from flask import request
import random

# 轮询负载均衡
def round_robin(services):
    services_list = list(services)
    return services_list[random.randint(0, len(services_list) - 1)]

@app.route('/recall')
def recall():
    # 选择服务
    service = round_robin(recall_services)

    # 转发请求
    response = requests.get(f"http://{service['Address']}:{service['Port']}/recall")
    return response.json()

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
def recall_with_circuit(user_id):
    """带熔断的召回"""
    return recall_items(user_id)

@app.route('/recall')
def recall():
    user_id = request.args.get('user_id')

    try:  # try/except捕获异常
        items = recall_with_circuit(user_id)
        return {'items': items}
    except Exception as e:
        # 降级:返回热门物品
        return {'items': get_hot_items()}

import logging
from logging.handlers import RotatingFileHandler

# 配置日志
logger = logging.getLogger('recommendation')
logger.setLevel(logging.INFO)

# 文件处理器
handler = RotatingFileHandler(
    'recommendation.log',
    maxBytes=10*1024*1024,
    backupCount=5
)
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)

# 记录日志
logger.info(f"Recall for user {user_id}: {len(items)} items")

def check_alerts():
    """检查告警"""
    # 检查QPS
    qps = get_current_qps()
    if qps < threshold_qps:
        send_alert(f"QPS too low: {qps}")

    # 检查延迟
    latency = get_avg_latency()
    if latency > threshold_latency:
        send_alert(f"Latency too high: {latency}ms")

    # 检查错误率
    error_rate = get_error_rate()
    if error_rate > threshold_error:
        send_alert(f"Error rate too high: {error_rate}%")

def send_alert(message):
    """发送告警"""
    # 发送邮件、短信、钉钉等
    pass

# API网关
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/recommend', methods=['GET'])
def recommend():
    """推荐接口"""
    user_id = request.args.get('user_id')
    context = request.args.to_dict()

    # 1. 召回
    recall_response = requests.get(
        f"http://recall-service/recall?user_id={user_id}"
    )
    candidates = recall_response.json()['items']

    # 2. 排序
    rank_response = requests.post(
        "http://rank-service/rank",
        json={
            'user_id': user_id,
            'items': candidates,
            'context': context
        }
    )
    ranked_items = rank_response.json()['items']

    # 3. 重排
    final_items = rerank_items(user_id, ranked_items)

    return jsonify({'items': final_items[:10]})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=80)