Spark文本相似度：Spark相似度计算如何实现
作者：张老师 浏览次数： 2020-02-19 16:52
在大数据平台系统当中，对于推荐系统、文本聚类、图像视频聚类等方面的计算需求，通常就涉及到相似度计算的相关问题。以文本聚类为例，如果在大数据计算当中遇到文本相似度计算的处理需求，应该如何去实现呢？今天，我们就主要来分享一下Spark文本相似度计算方面的知识。
如上所言，大规模相似度计算的需求，常常在推荐系统、文本聚类以及图像视频聚类当中出现，而文本聚类，也就是文本相似度计算，相对推荐系统、图像聚类来说，实现难度不算太高。
在Spark框架当中，实现相似度计算，可以基于笛卡尔积的方式计算RDD中两两匹配的相似度。
RDD作为分布式计算数据集，可以很容易的实现多线程、多节点的进行相似度计算，比单机单进程的计算方式要快很多。一般业务场景下item数量要远远大于user数据量。所以，在大数据量下，即时单机可以容忍itemCF的计算时间，userCF则不行。最简单的RDD优化方式，可以通过广播变量的方式提高对比计算的效率。
向量化+分布式矩阵
RowMatrix
CoordinateMatrix
//利用Spark分布式矩阵计算Item相似度
def itemVectorSimilarity(spark:SparkSession,
trainData:Dataset[UserItem],
topN: Int) = {
//计算相似度矩阵
def standardCosine(matrix: CoordinateMatrix): RDD[MatrixEntry] = {
val similarity= matrix.toIndexedRowMatrix().columnSimilarities()
val sim= similarity.entries
sim
}
val sim= standardCosine(parseToMatrix(trainData)).map {
case MatrixEntry(user1, user2, sim)=> (user1, (user2,sim))
}.groupByKey().
flatMapValues {x=>
val sim_users = x.toList.sortBy(-_._2).take(topN)
sim_users
}.map(x=> (x._1, x._2._1,x._2._2))
}
LSH
Spark实现LSH（局部敏感哈希）如：BucketedRandomProjectionLSH 和 MinHash。
LSH保证高维空间相近的点映射到低维空间相近的概率很高。LSH也是一个降维的过程。
LSH的参数
Tables：该参数越大，可以降低误报率（false negative rate）；参数越小，可以提高运行的效率。
K-Means Tree
K-Means Tree实际就是对数据做了多层K-means，每一层到当前的划分“叶子节点”包含样本数都小于1个。
K-D Tree
是对数据点在k维空间中划分的一种数据结构。k-d tree实际上是一种二叉树。
Spark-LSH实现Item相似度计算：
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.linalg.{SparseVector,Vectors}
import org.apache.spark.ml.feature.{BucketedRandomProjectionLSH, MinHashLSH, MinHashLSHModel}
val user_count =data_rating.map(_._1.toInt).max
val data_rating_matrix = data_rating.map (x=> (x._2.toInt, (x._1.toInt, x._3))).
groupByKey().mapValues { x =>
val features= x.toList.map {
case (feature_index, value) =>
(feature_index, value)
}.sortBy(_._1)
Vectors.sparse(user_count - 1, features)
}.sortByKey()
val data_rating_df =spark.createDataFrame(data_rating_matrix).toDF("id", "features")
val mh = new MinHashLSH().
setNumHashTables(5).
setInputCol("features").
setOutputCol("hashes")
val itemModel = mh.fit(data_rating_df)
val transformed_data_rating =itemModel.transform(data_rating_df)
transformed_data_rating.show(false)
val itemSimRDD = data_rating_matrix.collect().
map(x=> (x._1, itemModel.approxNearestNeighbors(data_rating_df, x._2, 10)))
itemSimRDD
目前来说，Spark文本相似度计算的最佳解决方案，是LSH +TopN，无需计算笛卡尔积，只要查找最近邻TopN即可。如果是计算笛卡尔积，时间复杂度至少 O（2^n）, 如果变成TopN查找问题，时间复杂度立马下降到O（n）。
关于Spark文本相似度计算，以上就是简单的介绍了，基于Spark在计算性能上的优势，大规模相似度计算的任务，可以快速地完成，并且将结果反馈到用户端。成都加米谷大数据，专业大数据培训机构，大数据开发培训每月开班中，课程详情可联系客服了解！
如上所言，大规模相似度计算的需求，常常在推荐系统、文本聚类以及图像视频聚类当中出现，而文本聚类，也就是文本相似度计算，相对推荐系统、图像聚类来说，实现难度不算太高。
在Spark框架当中，实现相似度计算，可以基于笛卡尔积的方式计算RDD中两两匹配的相似度。
RDD作为分布式计算数据集，可以很容易的实现多线程、多节点的进行相似度计算，比单机单进程的计算方式要快很多。一般业务场景下item数量要远远大于user数据量。所以，在大数据量下，即时单机可以容忍itemCF的计算时间，userCF则不行。最简单的RDD优化方式，可以通过广播变量的方式提高对比计算的效率。
向量化+分布式矩阵
RowMatrix
CoordinateMatrix
//利用Spark分布式矩阵计算Item相似度
def itemVectorSimilarity(spark:SparkSession,
trainData:Dataset[UserItem],
topN: Int) = {
//计算相似度矩阵
def standardCosine(matrix: CoordinateMatrix): RDD[MatrixEntry] = {
val similarity= matrix.toIndexedRowMatrix().columnSimilarities()
val sim= similarity.entries
sim
}
val sim= standardCosine(parseToMatrix(trainData)).map {
case MatrixEntry(user1, user2, sim)=> (user1, (user2,sim))
}.groupByKey().
flatMapValues {x=>
val sim_users = x.toList.sortBy(-_._2).take(topN)
sim_users
}.map(x=> (x._1, x._2._1,x._2._2))
}
LSH
Spark实现LSH（局部敏感哈希）如：BucketedRandomProjectionLSH 和 MinHash。
LSH保证高维空间相近的点映射到低维空间相近的概率很高。LSH也是一个降维的过程。
LSH的参数
Tables：该参数越大，可以降低误报率（false negative rate）；参数越小，可以提高运行的效率。
K-Means Tree
K-Means Tree实际就是对数据做了多层K-means，每一层到当前的划分“叶子节点”包含样本数都小于1个。
K-D Tree
是对数据点在k维空间中划分的一种数据结构。k-d tree实际上是一种二叉树。
Spark-LSH实现Item相似度计算：
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.linalg.{SparseVector,Vectors}
import org.apache.spark.ml.feature.{BucketedRandomProjectionLSH, MinHashLSH, MinHashLSHModel}
val user_count =data_rating.map(_._1.toInt).max
val data_rating_matrix = data_rating.map (x=> (x._2.toInt, (x._1.toInt, x._3))).
groupByKey().mapValues { x =>
val features= x.toList.map {
case (feature_index, value) =>
(feature_index, value)
}.sortBy(_._1)
Vectors.sparse(user_count - 1, features)
}.sortByKey()
val data_rating_df =spark.createDataFrame(data_rating_matrix).toDF("id", "features")
val mh = new MinHashLSH().
setNumHashTables(5).
setInputCol("features").
setOutputCol("hashes")
val itemModel = mh.fit(data_rating_df)
val transformed_data_rating =itemModel.transform(data_rating_df)
transformed_data_rating.show(false)
val itemSimRDD = data_rating_matrix.collect().
map(x=> (x._1, itemModel.approxNearestNeighbors(data_rating_df, x._2, 10)))
itemSimRDD
目前来说，Spark文本相似度计算的最佳解决方案，是LSH +TopN，无需计算笛卡尔积，只要查找最近邻TopN即可。如果是计算笛卡尔积，时间复杂度至少 O（2^n）, 如果变成TopN查找问题，时间复杂度立马下降到O（n）。
关于Spark文本相似度计算，以上就是简单的介绍了，基于Spark在计算性能上的优势，大规模相似度计算的任务，可以快速地完成，并且将结果反馈到用户端。成都加米谷大数据，专业大数据培训机构，大数据开发培训每月开班中，课程详情可联系客服了解！
