2024年电商用户行为数据集解析：从数据清洗到特征工程实战指南

1次阅读

共计 2648 个字符，预计需要花费 7 分钟才能阅读完成。

电商用户行为数据通常包含点击流、加购、下单等多类型事件，具有明显的时间序列特性（如用户从浏览到购买的路径）。这类数据往往存在大量非结构化日志（如 JSON 格式的埋点数据），且用户行为间存在时空关联性（例如同一会话中的连续操作）。数据维度高但稀疏（如商品 ID 的 One-Hot Encoding/ 独热编码）是其典型特征。

原始数据通常为嵌套 JSON 格式，需要处理如下情况：

import json
import pandas as pd

# 示例日志结构
log_example = '''{"user_id":"U123","events": [{"type": "click", "timestamp": "2024-01-01T12:00:00", "item_id": "I456"},
    {"type": "purchase", "timestamp": "2024-01-01T12:05:00", "order_id": "O789"}
  ]
}
'''

# 展开嵌套结构
data = json.loads(log_example)
rows = []
for event in data['events']:
    row = {'user_id': data['user_id']}
    row.update(event)
    rows.append(row)

df = pd.DataFrame(rows)

用户行为序列的时空关系可通过 Session 划分和转移概率矩阵表示。常用 Session 切割规则：

30 分钟内无操作视为新会话
跨天的行为自动分界

计算转移概率的 LaTeX 公式：

$$P(a\rightarrow b) = \frac{N(a\rightarrow b)}{\sum_{x}N(a\rightarrow x)}$$

当商品 ID 超过百万时，可采用如下优化方案：

Hash Bucketing/ 哈希分桶：df['item_hash'] = df['item_id'].apply(lambda x: hash(x) % 10000)
频次过滤：剔除出现次数 <5 的长尾商品
内存映射：pd.DataFrame.sparse.from_spmatrix()

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ecommerce_analysis") \
    .config("spark.executor.memory", "8g") \
    .config("spark.driver.memory", "4g") \
    .getOrCreate()

# 读取 TB 级数据
df = spark.read.parquet("hdfs://path/to/data/*.parquet") \
    .repartition(1000)  # 按日期分区

# UV/PV 统计（注意避免逐行循环）def get_metrics(df):
    return pd.DataFrame({'date': pd.to_datetime(df['timestamp']).dt.date,
        'user_id': df['user_id'],
        'is_click': (df['event_type'] == 'click').astype(int)
    }).groupby('date').agg(UV=('user_id', 'nunique'),  # 唯一用户数
        PV=('is_click', 'sum')      # 点击总量
    )

from sklearn.preprocessing import QuantileTransformer

# 计算最近购买日（Recency）current_date = pd.to_datetime('2024-06-01')
rfm = df[df['event_type'] == 'purchase'].groupby('user_id').agg(R=('timestamp', lambda x: (current_date - x.max()).days),
    F=('order_id', 'count'),
    M=('order_amount', 'sum')
)

# 分箱归一化
scaler = QuantileTransformer(n_quantiles=10)
rfm[['R_score','F_score','M_score']] = scaler.fit_transform(rfm)
rfm['RFM_score'] = rfm[['R_score','F_score','M_score']].mean(axis=1)

分区方式	1TB 数据读取时间
无分区	48min
按日期分区	12min
按小时 + 用户 ID	8min

 原始数据：- 尺寸：120GB 
- 内存：38GB（加载后）特征矩阵：- 尺寸：4.7GB（稀疏存储）- 内存：6.2GB

# 统一转换为 UTC+8
df['timestamp'] = pd.to_datetime(df['timestamp']) \
    .dt.tz_localize('UTC') \
    .dt.tz_convert('Asia/Shanghai')

使用 SHA256 替代原生 hash：import hashlib; hashlib.sha256(user_id.encode()).hexdigest()
维护映射表校验碰撞

# 计算每周统计量差异
from scipy.stats import wasserstein_distance

def check_drift(current, historical):
    return {'click_dist': wasserstein_distance(current['click_rate'], historical['click_rate']),
        'price_dist': wasserstein_distance(current['avg_price'], historical['avg_price'])
    }