推荐系统实战：如何用余弦相似度找到相似用户（含Spark优化技巧）

1. 余弦相似度的工业级应用价值

在推荐系统领域，找到相似用户是构建个性化推荐的基础。想象一下，当你在电商平台浏览商品时，系统能够精准推荐"和你品味相似的用户也喜欢"的商品，这种体验背后往往就隐藏着余弦相似度的数学魔法。

与传统教材中偏重公式推导不同，工业场景更关注如何高效处理海量数据。以MovieLens数据集为例，当用户规模达到千万级，物品数量突破百万时，简单的矩阵计算就会面临严峻挑战：

• 稀疏性问题：用户-物品评分矩阵通常99%以上都是零值
• 计算复杂度：传统算法的时间复杂度可能达到O(n³)量级
• 实时性要求：线上服务需要毫秒级响应推荐请求

# 典型用户-物品矩阵的稀疏结构示例
import numpy as np
user_item_matrix = np.array([
    [5, 0, 0, 1, 0],  # 用户1
    [0, 4, 3, 0, 0],  # 用户2
    [1, 0, 0, 0, 5],  # 用户3
])
print(f"稀疏度: {1 - np.count_nonzero(user_item_matrix)/user_item_matrix.size:.1%}")

2. 余弦相似度的工程化实现

2.1 基础公式的优化变形

原始余弦相似度公式：

$$ \text{similarity} = \frac{A \cdot B}{||A|| \times ||B||} $$

在实际工程中，我们通常会进行以下优化：

1. 向量归一化预处理：提前计算并存储向量的模长
2. 稀疏矩阵压缩：使用CSR/CSC格式存储非零元素
3. 并行点积计算：将向量分块进行分布式计算

// Spark中归一化余弦相似度计算示例
def cosineSimilarity(vec1: Vector, vec2: Vector): Double = {
  val dotProduct = vec1.toArray.zip(vec2.toArray).map(p => p._1 * p._2).sum
  val norm1 = math.sqrt(vec1.toArray.map(x => x * x).sum)
  val norm2 = math.sqrt(vec2.toArray.map(x => x * x).sum)
  dotProduct / (norm1 * norm2)
}

2.2 相似度计算的四大陷阱

问题类型	现象	解决方案
零向量问题	模长为零导致除零错误	添加微小epsilon值
长尾分布	热门物品主导相似度	TF-IDF加权
维度诅咒	高维空间距离失效	降维处理
冷启动	新用户/物品数据不足	混合推荐策略

提示：在实际应用中，余弦相似度值小于0.05通常可以视为不相关，大于0.7则认为高度相似

3. Spark分布式优化实战

3.1 基于ALS的矩阵分解

交替最小二乘法(ALS)是处理稀疏矩阵的利器：

1. 将原始矩阵分解为低维用户矩阵和物品矩阵
2. 相似度计算在低维空间进行
3. 天然支持分布式计算

from pyspark.ml.recommendation import ALS

als = ALS(
    rank=50,           # 潜在因子数量
    maxIter=20,        # 迭代次数
    regParam=0.01,     # 正则化参数
    implicitPrefs=True # 隐式反馈
)
model = als.fit(ratings)

3.2 基于MinHash的近似计算

当数据规模达到亿级时，精确计算变得不现实。MinHash算法可以在损失少量精度的情况下大幅提升性能：

1. 使用多个哈希函数生成签名
2. 通过签名相似度估计原始集合相似度
3. 计算复杂度从O(n²)降到O(n)

import org.apache.spark.ml.feature.MinHashLSH

val mh = new MinHashLSH()
  .setNumHashTables(5)
  .setInputCol("features")
  .setOutputCol("hashes")

val model = mh.fit(df)
model.approxSimilarityJoin(df, df, 0.6)

4. 性能调优与监控

4.1 关键性能指标

指标	合理范围	监控方法
计算耗时	<500ms	Spark UI
内存使用	<70% Executor内存	Ganglia
网络IO	<1Gbps	NetData
相似度分布	峰值0.3-0.7	Histogram

4.2 常见优化手段

• 数据倾斜处理：

  # 使用salting技术解决倾斜问题
  df = df.withColumn("salt", (rand() * 100).cast("int"))
  

• 缓存策略选择：

  ratings.persist(StorageLevel.MEMORY_AND_DISK_SER)
  

• 参数调优：

  spark-submit --num-executors 100 \
               --executor-cores 4 \
               --executor-memory 16g
  

在实际项目中，我们曾通过以下组合将计算时间从6小时缩短到15分钟：

1. 使用Parquet格式存储替代CSV
2. 将Spark的shuffle分区数从200调整到1000
3. 对用户向量采用广播变量优化

5. 业务场景中的创新应用

5.1 动态兴趣衰减模型

传统余弦相似度计算忽略时间因素，改进方案：

def time_decay(score, timestamp):
    half_life = 30 * 86400  # 30天半衰期
    return score * 0.5 ** ((current_time - timestamp)/half_life)

5.2 多维度融合相似度

结合多种行为数据计算综合相似度：

行为类型	权重	归一化方法
浏览	0.3	点击次数/log(总点击)
收藏	0.5	布尔值
购买	0.8	金额分箱

5.3 图结构扩展

将用户相似度转化为图结构，应用社区发现算法：

import org.apache.spark.graphx.Graph

val userGraph = Graph.fromEdges(
  similarityEdges.rdd.map(r => 
    Edge(r.getLong(0), r.getLong(1), r.getDouble(2))
  ), 0.0
)

在大型电商平台的实践中，这种方案使推荐点击率提升了12%。