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

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天后仍保留50%影响
• 衰减曲线采用指数形式，符合记忆规律

2.2 序列特征进阶处理

2.2.1 会话分割技术

用户行为流需要合理切分为会话(Session)。以下是基于超时阈值的实现：

python复制from pyspark.sql.window import Window
import pyspark.sql.functions as F

def sessionize_logs(logs_df, timeout_mins=30):
    window = Window.partitionBy("user_id").orderBy("timestamp")
    
    # 计算相邻行为时间差
    logs_df = logs_df.withColumn(
        "time_diff", 
        unix_timestamp("timestamp") - 
        unix_timestamp(F.lag("timestamp", 1).over(window))
    )
    
    # 标识会话边界
    logs_df = logs_df.withColumn(
        "new_session",
        F.when(F.col("time_diff") > (timeout_mins * 60), 1).otherwise(0)
    )
    
    # 生成会话ID
    logs_df = logs_df.withColumn(
        "session_id",
        F.sum("new_session").over(window.rowsBetween(Window.unboundedPreceding, 0))
    )
    
    return logs_df

业务考量：

• 30分钟阈值基于电商用户行为分析：
  • 手机端平均单次使用时长约8-12分钟
  • 超过30分钟间隔可视为新会话
• 可扩展为动态阈值（如夜间时段延长至2小时）

2.2.2 序列Embedding生成

使用Transformer处理行为序列的简化示例：

python复制import tensorflow as tf
from tensorflow.keras.layers import Input, Embedding, TransformerEncoder

def build_sequence_model(vocab_size=10000, seq_length=50):
    # 输入层
    item_input = Input(shape=(seq_length,), dtype='int32')
    
    # Embedding层
    embedding = Embedding(
        input_dim=vocab_size,
        output_dim=64,
        input_length=seq_length
    )(item_input)
    
    # Transformer编码器
    encoder = TransformerEncoder(
        num_heads=4,
        dense_dim=128,
        dropout=0.1
    )(embedding)
    
    # 池化操作
    pooled = tf.reduce_mean(encoder, axis=1)
    
    return tf.keras.Model(inputs=item_input, outputs=pooled)

关键配置说明：

• Embedding维度64：平衡表达能力与计算成本
• 4个注意力头：捕捉不同子空间的序列模式
• 均值池化：保留整体序列特征，避免位置偏差

3. 大规模日志的工程实践

3.1 实时处理架构设计

现代推荐系统需要实时响应用户行为。以下是经过验证的Lambda架构：

code复制┌─────────────────┐    ┌─────────────────┐
│  实时流处理层    │    │  批量处理层     │
│  (Flink/Kafka)  │◄──►│ (Spark/Hadoop)  │
└────────┬────────┘    └────────┬────────┘
         │                       │
         ▼                       ▼
┌───────────────────────────────┐
│        统一服务层              │
│ (特征存储+模型服务)            │
└───────────────────────────────┘

组件选型对比：

3.2 数据质量保障体系

3.2.1 异常检测规则库

建立多维度检测规则：

python复制# 规则1：异常高频访问检测
abnormal_activity = logs.groupBy("user_id", "ip_address") \
    .agg(F.count("*").alias("action_count")) \
    .filter("action_count > 100")  # 阈值根据业务调整

# 规则2：非工作时间访问检测
night_activity = logs.filter(
    (F.hour("timestamp") < 6) | 
    (F.hour("timestamp") > 23)
).groupBy("user_id").count()

# 规则3：机器行为模式识别
bot_pattern = logs.groupBy("user_id", "session_id") \
    .agg(
        F.countDistinct("item_id").alias("unique_items"),
        F.avg("dwell_time").alias("avg_view_time")
    ).filter(
        (F.col("unique_items") > 50) & 
        (F.col("avg_view_time") < 2)
    )

3.2.2 数据一致性检查

实施端到端验证：

python复制def validate_data_pipeline(raw_logs, processed_logs):
    # 基数校验
    raw_count = raw_logs.count()
    processed_count = processed_logs.count()
    loss_rate = (raw_count - processed_count) / raw_count
    
    # 关键字段完整性
    null_checks = {
        'user_id': 0,
        'timestamp': 0.001,  # 允许0.1%缺失
        'action_type': 0
    }
    
    results = {}
    for field, threshold in null_checks.items():
        null_rate = processed_logs.filter(F.col(field).isNull()).count() / processed_count
        results[field] = null_rate <= threshold
    
    return {
        'total_loss_rate': loss_rate,
        'field_validation': results
    }

4. 推荐系统的伦理思考

4.1 多样性保障机制

避免信息茧房的工程技术方案：

python复制def diversity_aware_ranking(candidate_items, user_profile, alpha=0.3):
    # 原始相关性分数
    relevance_scores = model.predict(candidate_items, user_profile)
    
    # 品类多样性计算
    categories = [item['category'] for item in candidate_items]
    category_dist = Counter(categories)
    diversity_scores = 1 / (np.array([category_dist[c] for c in categories]) + 1)
    
    # 混合排序
    combined_scores = alpha * diversity_scores + (1 - alpha) * relevance_scores
    ranked_indices = np.argsort(combined_scores)[::-1]
    
    return [candidate_items[i] for i in ranked_indices]

参数说明：

• α=0.3：经验值，平衡相关性与多样性
• 逆频率加权：提升长尾品类曝光机会

4.2 探索-利用平衡

实现ε-greedy策略的工程化方案：

python复制class ExploreExploitRouter:
    def __init__(self, initial_epsilon=0.1, decay_rate=0.999):
        self.epsilon = initial_epsilon
        self.decay_rate = decay_rate
    
    def route(self, user_id, main_recs, explore_pool):
        if random.random() < self.epsilon:
            # 探索阶段
            selected = random.sample(explore_pool, k=min(3, len(explore_pool)))
            self.epsilon *= self.decay_rate  # 衰减探索率
            return {
                'type': 'explore',
                'items': selected
            }
        else:
            # 利用阶段
            return {
                'type': 'exploit',
                'items': main_recs[:5]
            }

动态调整策略：

• 新用户：ε=0.3，快速探索兴趣
• 活跃用户：ε=0.05，保持少量探索
• 衰减机制：随着交互次数增加逐渐降低探索比例

5. 实战经验与避坑指南

5.1 时间戳处理陷阱

常见问题：

• 多时区日志混合（服务器时间 vs 客户端时间）
• 夏令时切换导致的时间跳变
• 设备时钟不同步产生的未来时间戳

解决方案：

python复制def normalize_timestamps(df, time_col='timestamp'):
    # 统一转换为UTC+8时区
    df = df.withColumn(
        'normalized_time',
        F.from_utc_timestamp(
            F.to_utc_timestamp(F.col(time_col), 'UTC'),
            'Asia/Shanghai'
        )
    )
    
    # 过滤异常时间点（未来时间或过早时间）
    df = df.filter(
        (F.col('normalized_time') < F.current_timestamp()) & 
        (F.col('normalized_time') > F.lit('2020-01-01'))
    )
    
    return df

5.2 冷启动处理策略

混合推荐方案：

1. 基于内容的过滤（Content-based）

   python复制def content_based_recommend(new_user, items, top_k=5):
       # 提取用户注册信息中的关键词
       interest_keywords = extract_keywords(new_user['profile'])
       
       # 计算物品内容匹配度
       scores = []
       for item in items:
           item_keywords = item['metadata']['keywords']
           overlap = len(set(interest_keywords) & set(item_keywords))
           scores.append(overlap / len(interest_keywords))
       
       # 返回Top-K推荐
       return sorted(zip(items, scores), key=lambda x: -x[1])[:top_k]
   

2. 热门榜单兜底

   python复制def get_popular_items(location=None, time_range='7d'):
       # 从特征存储获取预计算的热门商品
       popular_df = feature_store.query(
           "popular_items",
           filter_expr=f"time_range='{time_range}' AND location='{location}'"
       )
       return popular_df.orderBy('score').limit(100).collect()
   

5.3 特征回填策略

处理日志延迟的工程方案：

python复制def backfill_features(feature_table, new_logs, time_field='event_time'):
    # 找出需要更新的时间范围
    min_new_time = new_logs.select(F.min(time_field)).first()[0]
    
    # 查询受影响的历史特征
    historical_features = feature_table.filter(
        F.col(time_field) >= min_new_time
    )
    
    # 重新计算特征
    updated_features = recompute_features(
        historical_features.union(new_logs)
    )
    
    # 原子性更新
    feature_table.delete_where(f"event_time >= '{min_new_time}'")
    feature_table.insert(updated_features)

6. 效能监控体系构建

6.1 指标监控大盘

核心监控指标配置示例：

python复制class RecommendationMonitor:
    metrics = {
        'daily_active_users': {
            'query': "SELECT COUNT(DISTINCT user_id) FROM logs WHERE dt='{date}'",
            'threshold': {'warn': 0.9, 'critical': 0.7}  # 同比变化阈值
        },
        'ctr': {
            'query': """
                SELECT 
                    SUM(CASE WHEN action='click' THEN 1 ELSE 0 END) / 
                    SUM(CASE WHEN action='impression' THEN 1 ELSE 0 END)
                FROM logs WHERE dt='{date}'
            """,
            'window': '7d'  # 计算滑动平均值
        },
        'diversity': {
            'query': """
                SELECT COUNT(DISTINCT category_id) / COUNT(*) 
                FROM rec_results WHERE dt='{date}'
            """
        }
    }
    
    def check_metrics(self, date):
        alerts = []
        for name, config in self.metrics.items():
            value = execute_query(config['query'].format(date=date))
            if 'threshold' in config:
                baseline = get_historical_avg(name, config.get('window'))
                ratio = value / baseline
                if ratio < config['threshold']['critical']:
                    alerts.append(f'CRITICAL: {name} dropped to {ratio:.2f}')
                elif ratio < config['threshold']['warn']:
                    alerts.append(f'WARNING: {name} dropped to {ratio:.2f}')
        return alerts

6.2 A/B测试框架

分层分流实现方案：

python复制class ABTestRouter:
    def __init__(self, experiments):
        self.experiments = experiments
        self.user_buckets = {}  # user_id -> bucket_mapping
        
    def assign_user(self, user_id):
        if user_id not in self.user_buckets:
            bucket = {}
            for exp in self.experiments:
                # 使用一致性哈希确保用户始终进入相同分组
                hash_val = int(hashlib.md5(f"{exp['name']}_{user_id}".encode()).hexdigest(), 16)
                bucket[exp['name']] = 'A' if (hash_val % 100) < exp['split'] else 'B'
            self.user_buckets[user_id] = bucket
        return self.user_buckets[user_id]
    
    def get_variant(self, user_id, experiment_name):
        return self.assign_user(user_id)[experiment_name]

7. 未来演进方向

7.1 多模态日志融合

下一代推荐系统需要整合：

• 浏览行为（页面滚动、鼠标轨迹）
• 视觉关注（眼动追踪、图片停留）
• 语音交互（语音搜索、反馈）

python复制class MultiModalLogger:
    def __init__(self):
        self.visual_attention = None
        self.voice_commands = []
    
    def log_eye_tracking(self, coordinates):
        # 记录视线焦点坐标
        self.visual_attention = self._cluster_gaze_points(coordinates)
        
    def log_voice_query(self, transcript, intent):
        self.voice_commands.append({
            'text': transcript,
            'intent': intent,
            'timestamp': time.time()
        })
    
    def get_engagement_score(self):
        # 综合多模态信号计算参与度
        visual_score = len(self.visual_attention) if self.visual_attention else 0
        voice_score = len([c for c in self.voice_commands if c['intent'] == 'purchase'])
        return 0.4 * visual_score + 0.6 * voice_score

7.2 因果推理应用

突破相关性局限，建立因果图模型：

python复制class CausalRecommender:
    def __init__(self, graph_config):
        self.graph = build_causal_graph(graph_config)
        
    def recommend(self, user_state, intervention=None):
        # 计算反事实结果
        baseline = self.graph.predict_outcome(user_state)
        if intervention:
            counterfactual = self.graph.predict_outcome(
                user_state, do=intervention
            )
            uplift = counterfactual - baseline
            return uplift
        return baseline

# 示例因果图配置
product_graph = {
    'nodes': ['price', 'reviews', 'promotion', 'conversion'],
    'edges': [
        ('price', 'conversion'),
        ('reviews', 'conversion'), 
        ('promotion', 'price'),
        ('promotion', 'conversion')
    ]
}

在日志分析这条路上，我越来越深刻体会到：技术手段再先进，也不能替代对用户行为的敬畏之心。每次看到团队沉迷于模型调参而忽视数据质量时，我都会想起那个经典比喻——"Garbage in, gospel out"（垃圾进，福音出）。真正的推荐系统专家，首先应该是个优秀的数据人类学家。