如何利用3-4-5规则数据挖掘解决高维稀疏数据特征提取难题

1次阅读

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

在电商用户行为分析、推荐系统等场景中，原始数据往往呈现高维稀疏特性。例如一个千万级用户的点击日志，经过 One-Hot 编码后可能产生数百万维的特征，但单个用户的非零特征占比通常不足 0.1%。这种数据结构会导致两个核心问题：

存储效率低下 ：稀疏矩阵的存储开销随维度呈指数增长，传统存储方式造成资源浪费
计算性能瓶颈 ：机器学习训练过程中的矩阵运算会因大量零值参与计算导致无效计算量激增

方法	保持原始特征	可解释性	计算复杂度
PCA	❌	弱	O(n³)
LDA	❌	中	O(n²k)
3-4- 5 规则	✅	强	O(n log n)

3 层特征分级
一级特征：用户显式行为（购买、收藏）
二级特征：隐式行为（浏览时长、页面滚动）
三级特征：环境特征（设备类型、网络环境）
4 类关联维度
用户维度：UserID、SessionID
时间维度：事件发生的时间片划分
空间维度：地理位置层级
物品维度：商品类目树
5 步过滤机制
频次过滤：剔除出现次数 < 阈值 τ 的特征
方差过滤：保留方差 >δ 的特征
互信息过滤：选择与目标变量互信息 >μ 的特征
相关性过滤：消除高度线性相关特征（ρ>0.9）
时效过滤：根据时间衰减因子 λ 淘汰陈旧特征

from pyspark.sql import functions as F

# 用户行为日志示例结构
behavior_df = spark.read.parquet("s3://bucket/user_behavior/")\
    .select(F.col("user_id").cast("string"),
        F.col("item_id").cast("int"),
        F.col("action_type"),  # 1:click, 2:cart, 3:purchase
        F.col("timestamp").cast("long")
    )

def stratified_sampling(df, strata_col, sample_ratio):
    """
    分层抽样实现
    :param df: 输入 DataFrame
    :param strata_col: 分层列名
    :param sample_ratio: 各层采样比例 dict
    :return: 采样后的 DataFrame
    """
    return df.sampleBy(
        strata_col,
        fractions=sample_ratio,
        seed=42
    )

# 按行为类型分层抽样
sampled_df = stratified_sampling(
    behavior_df,
    "action_type",
    {1: 0.1, 2: 0.3, 3: 0.5}  # 对稀有行为过采样
)

from pyspark.ml.feature import CountVectorizer

# 时间衰减权重计算
def time_decay_weight(timestamp, half_life=30*24*3600):
    """
    指数衰减权重计算
    :param timestamp: 事件时间戳
    :param half_life: 半衰期（秒）:return: 衰减权重值
    """
    current_time = F.unix_timestamp()
    delta = (current_time - timestamp) / half_life
    return F.exp(-delta * F.lit(0.6931))  # ln(2)

weighted_df = sampled_df.withColumn(
    "weight",
    time_decay_weight("timestamp")
)

识别倾斜键

# 找出频次最高的 100 个 item_id
skew_keys = behavior_df.groupBy("item_id")\
    .count()\
    .orderBy(F.desc("count"))\
    .limit(100)\
    .rdd.map(lambda x: x[0])\
    .collect()

倾斜处理

from pyspark.sql import Window

# 对热门商品进行分桶处理
window_spec = Window.partitionBy("item_id").orderBy("user_id")
bucketed_df = behavior_df.withColumn(
    "bucket_id",
    F.when(F.col("item_id").isin(skew_keys),
        F.floor(F.rand(42) * 100)  # 随机分 100 桶
    ).otherwise(F.lit(0))
)

# 关键 Spark 配置
spark.conf.set("spark.sql.shuffle.partitions", "2000")  # 根据集群规模调整
spark.conf.set("spark.sql.adaptive.enabled", "true")  # 启用 AQE
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

特征级匿名化
用户 ID 通过 HMAC-SHA256 加密
地理位置模糊到城市级别
IP 地址保留前 24 位

差分隐私实现

import numpy as np

def add_laplace_noise(values, epsilon=0.1):
    """
    添加拉普拉斯噪声
    :param values: 原始值数组
    :param epsilon: 隐私预算
    :return: 加噪后的数组
    """
    scale = 1.0 / epsilon
    noise = np.random.laplace(0, scale, len(values))
    return values + noise

该规则可迁移到 IoT 时序数据分析场景：