Spark Python:如何计算RDD中每行之间的Jaccard相似性?

Question

>我有一个包含大约 50k 个不同行和 2 列的表格。您可以将每一行视为一部电影，列是该电影的属性 - "ID"：该电影的 ID，"标签"：电影的一些内容标签，以每部电影的字符串列表的形式出现。

数据如下所示：

movie_1，["浪漫"，"喜剧"，"英语"]; movie_2，["动作"，"功夫"，"中国"]

我的目标是首先根据每部电影的相应标签计算每部电影之间的提花相似性，一旦完成，我将能够知道每部电影(例如我选择 movie_1(，其他前5部最相似的电影是什么(在这种情况下movie_1(。我不仅希望获得movie_1本身的前5名结果，而且希望获得所有电影的前5名。

我尝试使用Python来解决问题，但是运行时在这里是一个很大的挑战。即使我使用多处理，在 6 个内核上运行，总运行时间仍然持续超过 20 小时。

下面的蟒蛇代码：

import pandas as pd
from collections import Counter
import numpy as np
from multiprocessing import Pool
import time
col_names=['movie_id','tag_name']
df=pd.read_csv("movies.csv",names=col_names)
movie_ids=df['movie_id'].tolist()
tag_list=df['tag_name'].tolist()
def jaccard_similarity(string1, string2):
intersection = set(string1).intersection(set(string2))
union = set(string1).union(set(string2))
return len(intersection)/float(len(union))
def jc_results(movie_id):
result=Counter()
this_index=movie_ids.index(movie_id)
for another_id in movie_ids:
that_index=movie_ids.index(another_id)
if another_id==movie_id:
continue
else:
tag_1=tag_list[this_index]
tag_2=tag_list[that_index]
jaccard = jaccard_similarity(tag_1,tag_2)
result[(movie_id,another_id)]=jaccard
return result.most_common(10)
from multiprocessing import Pool
pool=Pool(6)
results={}
for movie_id in movie_ids:
results[movie_id]=pool.apply_async(jc_results,args=(movie_id,))
pool.close()
pool.join()
for movie_id, res in results.items():
results[movie_id] = res.get()

然后我想切换到 Pyspark，但是我对 Spark python 仍然很陌生，并且在用它写了几行后卡住了，实际上我所做的唯一进步是使用 sc.textFile 将数据读取到 RDD...已经阅读了现有的帖子，但他们都在使用Scala。如果有人可以帮助或提供任何指导，那就太好了 Pyspark.多谢！

Answer 1

您可以尝试类似于此 stackoverflow 答案的解决方案，但由于您的数据已经被标记化(字符串列表(，因此您无需执行该步骤或 ngram 步骤。

我还要提到，pyspark 中的 approxSimilarityJoin 计算的是 Jaccard 距离而不是 Jaccard Similarity，但如果你特别需要，你可以从 1 中减去以转换回相似性。

您的代码最终将类似于：

from pyspark.ml import Pipeline
from pyspark.ml.feature import HashingTF, MinHashLSH
import pyspark.sql.functions as f
db = spark.createDataFrame([
('movie_1', ['romantic','comedy','English']),
('movie_2', ['action','kongfu','Chinese']),
('movie_3', ['romantic', 'action'])
], ['movie_id', 'genres'])

model = Pipeline(stages=[
HashingTF(inputCol="genres", outputCol="vectors"),
MinHashLSH(inputCol="vectors", outputCol="lsh", numHashTables=10)
]).fit(db)
db_hashed = model.transform(db)
db_matches = model.stages[-1].approxSimilarityJoin(db_hashed, db_hashed, 0.9)
#show all matches (including duplicates)
db_matches.select(f.col('datasetA.movie_id').alias('movie_id_A'),
f.col('datasetB.movie_id').alias('movie_id_B'),
f.col('distCol')).show()
#show non-duplicate matches
db_matches.select(f.col('datasetA.movie_id').alias('movie_id_A'),
f.col('datasetB.movie_id').alias('movie_id_B'),
f.col('distCol')).filter('movie_id_A < movie_id_B').show()

使用相应的输出：

+----------+----------+-------+
|movie_id_A|movie_id_B|distCol|
+----------+----------+-------+
|   movie_3|   movie_3|    0.0|
|   movie_1|   movie_3|   0.75|
|   movie_2|   movie_3|   0.75|
|   movie_1|   movie_1|    0.0|
|   movie_2|   movie_2|    0.0|
|   movie_3|   movie_2|   0.75|
|   movie_3|   movie_1|   0.75|
+----------+----------+-------+
+----------+----------+-------+
|movie_id_A|movie_id_B|distCol|
+----------+----------+-------+
|   movie_1|   movie_3|   0.75|
|   movie_2|   movie_3|   0.75|
+----------+----------+-------+