大规模推荐系统¶

# 使用PyTorch DistributedDataParallel
import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP

def setup_distributed():
    """初始化分布式环境"""
    dist.init_process_group(backend='nccl')
    local_rank = int(os.environ['LOCAL_RANK'])
    torch.cuda.set_device(local_rank)

def train_distributed(model, dataloader, rank):
    """分布式训练"""
    # 包装模型
    model = DDP(model, device_ids=[rank])

    # 训练循环
    for epoch in range(num_epochs):
        for batch in dataloader:
            # 前向传播
            optimizer.zero_grad()  # 清零梯度
            output = model(batch)
            loss = criterion(output, labels)

            # 反向传播
            loss.backward()  # 反向传播计算梯度
            optimizer.step()  # 更新参数

# 使用TensorFlow Parameter Server Strategy
import tensorflow as tf

strategy = tf.distribute.ParameterServerStrategy(
    cluster_resolver=tf.distribute.cluster_resolver.TFConfigClusterResolver()
)

with strategy.scope():
    model = build_model()
    optimizer = tf.keras.optimizers.Adam()

    model.compile(optimizer=optimizer, loss='binary_crossentropy')
    model.fit(train_dataset, epochs=10)

# 使用NCCL AllReduce
import torch
import torch.distributed as dist

def all_reduce(tensor):
    """AllReduce操作"""
    dist.all_reduce(tensor, op=dist.ReduceOp.SUM)
    tensor.div_(dist.get_world_size())
    return tensor

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

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

# 读取Kafka数据
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user_events") \
    .load()

# 特征计算
features = df \
    .select(
        col("user_id"),
        col("item_id"),
        col("event_type")
    ) \
    .groupBy("user_id") \
    .agg(
        count("event_type").alias("event_count"),
        avg("event_type").alias("avg_event")
    )

# 写入特征存储
query = features.writeStream \
    .format("redis") \
    .option("host", "localhost") \
    .option("port", "6379") \
    .start()

query.awaitTermination()

from flask import Flask, request
import redis
import hbase

app = Flask(__name__)

# 连接存储
redis_client = redis.Redis(host='localhost', port=6379)
hbase_client = hbase.ConnectionPool('localhost')

@app.route('/features', methods=['GET'])
def get_features():
    """获取特征"""
    user_id = request.args.get('user_id')

    # 从Redis获取热数据
    hot_features = redis_client.hgetall(f'user:{user_id}')

    # 从HBase获取温数据
    cold_features = hbase_client.get('user_features', user_id)

    # 合并特征
    features = {**hot_features, **cold_features}

    return features

if __name__ == '__main__':
    app.run(port=5000)

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# 创建Kafka源
t_env.execute_sql("""
    CREATE TABLE user_events (
        user_id BIGINT,
        item_id BIGINT,
        event_type STRING,
        timestamp TIMESTAMP(3)
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user_events',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
""")

# 实时计算
t_env.execute_sql("""
    INSERT INTO recommendations
    SELECT
        user_id,
        COLLECT_LIST(item_id) as items
    FROM (
        SELECT
            user_id,
            item_id,
            ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY timestamp DESC) as rn
        FROM user_events
    )
    WHERE rn <= 10
    GROUP BY user_id
""")

# 执行
env.execute("Real-time Recommendation")

class RealTimeRecommender:
    def __init__(self, model, feature_store):
        self.model = model
        self.feature_store = feature_store

    def recommend(self, user_id):
        """实时推荐"""
        # 1. 获取用户特征
        user_features = self.feature_store.get_user_features(user_id)

        # 2. 召回
        candidate_items = self.recall(user_features)

        # 3. 排序
        ranked_items = self.rank(user_features, candidate_items)

        # 4. 重排
        final_items = self.rerank(user_features, ranked_items)

        return final_items

    def recall(self, user_features):
        """召回"""
        # 从Redis获取候选物品
        return self.feature_store.get_candidates(user_features['user_id'])

    def rank(self, user_features, items):
        """排序"""
        scores = []
        for item_id in items:
            item_features = self.feature_store.get_item_features(item_id)
            score = self.model.predict(user_features, item_features)
            scores.append((item_id, score))

        # 排序
        scores.sort(key=lambda x: x[1], reverse=True)  # lambda匿名函数
        return [item_id for item_id, score in scores[:100]]  # 切片操作，取前n个元素

    def rerank(self, user_features, items):
        """重排"""
        # 多样性重排
        return self.diversify(items)

# 召回服务
from flask import Flask, request
import redis

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')

    # 多路召回
    cf_items = collaborative_filtering_recall(user_id)
    content_items = content_based_recall(user_id)
    hot_items = hot_items_recall()

    # 融合
    all_items = cf_items + content_items + hot_items

    # 去重
    unique_items = list(set(all_items))

    return {'items': unique_items[:500]}

# 排序服务
@app.route('/rank', methods=['POST'])
def rank():
    """排序接口"""
    data = request.json
    user_id = data['user_id']
    items = data['items']

    # 获取特征
    user_features = get_user_features(user_id)
    item_features_list = [get_item_features(item_id) for item_id in items]

    # 批量预测
    scores = model.predict_batch(user_features, item_features_list)

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

    return {'items': [item for item, score in ranked_items[:100]]}

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