推荐系统中日志数据处理与特征工程实战

1. 日志数据的价值再发现

在推荐系统领域，我们常常陷入一个认知误区：认为模型复杂度等同于推荐效果。从业十年，我见过太多团队将90%的精力投入在模型调参上，却对日志数据质量视而不见。这就像厨师执着于研究菜谱，却对食材新鲜度漠不关心——最终呈现的菜品必然大打折扣。

1.1 行为日志的本质解析

用户行为日志远不止是数据库里的几行记录，它们是用户在数字世界的"生物特征"。当我们观察以下日志片段时：

plaintext复制user_123 | product_A | view      | 2023-07-20 09:15:23
user_123 | product_A | click     | 2023-07-20 09:15:35
user_123 | product_B | view      | 2023-07-20 09:16:02
user_456 | product_C | add_to_cart | 2023-07-20 09:16:15

这些数据实际上构成了用户的"数字DNA"：

• 时间维度：行为间隔反映决策速度（12秒点击 vs 27秒跳过）
• 序列模式：view→click→add_to_cart形成转化漏斗
• 跨商品关联：A和B的浏览顺序暗示品类偏好

关键认知：高质量日志=用户心理活动的数字化投影。没有精准投影，任何模型都只是对空气挥拳。

1.2 推荐系统的数据金字塔

在我的实践经验中，推荐系统的效能遵循以下金字塔结构（效能影响权重）：

code复制       ▲
      / \
     /   \
    /模型层\ 20%
   /-------\
  /特征工程\ 30%
 /---------\
/ 数据质量 \ 50%
------------

这个结构揭示了残酷的现实：底层数据质量问题会以乘积效应向上传导。举例来说：

• 10%的曝光日志丢失 → 特征计算偏差放大到25% → 模型效果下降40%
• 时间戳未统一时区 → 时间衰减特征失真 → 用户兴趣漂移检测失效

2. 从原始日志到行为特征

2.1 基础特征工程实战

2.1.1 统计特征构造

以下是用PySpark实现的核心统计特征计算框架：

python复制from pyspark.sql import functions as F

def build_basic_features(logs_df):
    # 行为类型权重映射
    action_weights = {
        'view': 0.2,
        'click': 0.5,
        'favorite': 0.8,
        'purchase': 1.0
    }
    
    # 生成特征表达式
    agg_exprs = []
    for action, weight in action_weights.items():
        agg_exprs.append(
            F.sum(F.when(F.col("action") == action, weight).otherwise(0))
            .alias(f"{action}_weighted_sum")
        )
    
    # 多维聚合
    feature_df = logs_df.groupBy("user_id", "category_id").agg(
        *agg_exprs,
        F.count("*").alias("total_actions"),
        F.max("timestamp").alias("last_action_time")
    )
    
    # 计算复合兴趣分
    feature_df = feature_df.withColumn(
        "interest_score",
        F.col("click_weighted_sum") * 0.6 + 
        F.col("favorite_weighted_sum") * 0.3 +
        F.col("purchase_weighted_sum") * 0.1
    )
    
    return feature_df

设计要点解析：

1. 差异化权重：不同行为类型对兴趣的表征强度不同（点击 vs 购买）
2. 分层聚合：先按(user, category)分组，避免维度爆炸
3. 可解释性：最终interest_score各组成部分权重明确

2.1.2 时间衰减策略

用户兴趣具有明显的时效性特征。我们通过指数衰减模型实现：

python复制from pyspark.sql.functions import unix_timestamp, lit, exp

def apply_time_decay(feature_df, current_time=None):
    if not current_time:
        current_time = F.current_timestamp()
    
    half_life_days = {
        'view': 3,    # 浏览行为半衰期短
        'click': 7,   # 点击中等
        'purchase': 30 # 购买行为影响持久
    }
    
    for action, days in half_life_days.items():
        feature_df = feature_df.withColumn(
            f"{action}_decayed",
            F.col(f"{action}_weighted_sum") * 
            exp(-(unix_timestamp(current_time) - 
                 unix_timestamp(F.col("last_action_time"))) / 
                (days * 24 * 3600))
        )
    
    return feature_df

参数设计原理：

• 半衰期设置基于用户行为研究：
  • 浏览行为3天后记忆度衰减50%
  • 购买行为30天后仍