使用 Spark 设置相似性连接

Question

我有两个文本文件，每行格式为（id，数字序列）。 我也有一个阈值。

文件 1 如下所示，在第一行中，0 是 id，其余是 一串数字。

0 1 4 5 6
1 2 3 6
2 4 5 6

同样，我有以下内容的文件 2。

0 1 4 6
1 2 5 6
2 3 5

我必须找到那些相似度值大于或等于阈值的行。相似度值可以计算为两条线的交点除以两条线的并集。例如，file1 的第 id-0 行有 seq 1,4,5,6，而 file2 的第 id-0 行有 1,4,6。它们的交集大小 = 3，联合大小 = 4。因此它们的相似度将是 3/4 = 0.75，大于阈值。

我已经编写了 python 代码来完成这项任务并尝试将其转换为 Scala。

    with open("file1.txt") as f1:
        content1 = f1.readlines()
    content1 = [x.strip() for x in content1]

    with open("file2.txt") as f2:
        content2 = f2.readlines()
    content2 = [x.strip() for x in content2]

    threshold = 0.5
    final_index_list_with_similarity = []

    for i in range(len(content1)):
        for j in range(len(content2)):
            index_content1 = content1[i][0]
            index_content2 = content2[j][0]
            s = set(content1[i][1:])
            t = set(content2[j][1:])
            intersect = s & t
            intersect_size = len(intersect) - 1
            union_size = len(s) + len(t) - intersect_size - 2 #substracting two because I am getting two extra space.
            similarity = intersect_size/union_size
            if similarity >= threshold:
                final_index_list_with_similarity.append((index_content1, index_content2, similarity))
    print(final_index_list_with_similarity)

输出：[('0', '0', 0.75), ('1', '1', 0.5), ('2', '0', 0.5), ('2', '1', 0.5)]

到目前为止，我在 scala 中尝试过的东西看起来像这样。

val inputFile1 = args(0)
val inputFile2 = args(1)
val threshold = args(2).toDouble
val ouputFolder = args(3)

val conf = new SparkConf().setAppName("SetSimJoin").setMaster("local")
val sc = new SparkContext(conf)

val lines1 = sc.textFile(inputFile1).flatMap(line => line.split("\n"))
val lines2 = sc.textFile(inputFile2).flatMap(line => line.split("\n"))

val elements1 = lines1.map { x => x.drop(x.split(" ")(0).length.toInt + 1) }.flatMap { x => x.split(" ") }.map { x => (x, 1) }.reduceByKey(_+_)
val elements2 = lines2.map { x => x.drop(x.split(" ")(0).length.toInt + 1) }.flatMap { x => x.split(" ") }.map { x => (x, 1) }.reduceByKey(_+_)

这给了我整个文件中每个数字的频率。

任何帮助或指导将不胜感激。

Answer 1

两个文件都可以合并为RDD，然后应用公式：“交集大小/联合大小”：

val lines1 = sparkContext.textFile("inputFile1")
val lines2 = sparkContext.textFile("inputFile2")

val rdd1 = lines1.map(_.split(" ")).map(v => (v(0), v.tail))
val rdd2 = lines2.map(_.split(" ")).map(v => (v(0), v.tail))

val result = rdd1.join(rdd2).map(r => (
  r._1,
  r._2._1.intersect(r._2._2).size * 1.0 /
    r._2._1.union(r._2._2).distinct.size
)
)

result.foreach(println)

输出是：

(1,0.5)
(0,0.75)
(2,0.25)