apache-spark - 使用 Apache-Spark，根据条件减少或折叠 RDD

Question

我正在使用 Apache Spark 和 Scala。我有一个 String,Int 的 RDD

val counts =words.map(word => (word, 1)).reduceByKey((a,b) => (a + b))     

现在我通过 Key 减少了 RDD，但我想添加另一个功能来减少相似的单词。

我想使用 Levenshtein 距离、欧氏距离或余弦距离。

那么，我该如何应用其中一个函数来减少我的 RDD？

示例:

RDD ->  (forks,12), (fork,4), (chair,15) , (table,1), (tables,11)

承认相似性算法有效，我怎样才能获得像这样的减少的 RDD:

RDD -> (fork,16), (table,12), (chair,15)

我试过类似的方法:

counts.foldLeft(){(x,y) => 
  if(x._1.euclideanDistance(y._1) > 0.9) 
    (x,x._2+y._2) 
}

Answer 1

你正在尝试的是行不通的。

如果你只有一个distance(a, b)函数，解决这个问题确实效率低下且复杂。您需要使用 RDD.cartesian 生成所有可能的 (word1, word2) 对。然后过滤掉那些距离太大的。现在你有了相似的词对。假设它们是 (fox, fix)、(fix, six) 以及它们的反转。然后，您想要对 fox、fix 和 six 的计数求和。为此，您需要在由相似词对定义的图中找到连通分量。获得每个单词的组件 ID 后，您可以按组件 ID 对计数求和。

我认为解决方案是编写一个可以将单词转换为其“规范”形式的函数。它会将 forks、forking 和 forked 变成 fork。然后你可以再次应用它和 reduceByKey。

如果没有Spark，这一步会最快。一旦您使用 Spark 计算了计数，您就有了一个很小的数据集——一个整数代表每个不同的单词。最简单的方法是收集，然后在本地map 和groupBy counts。