对 approxSimilarityJoin 进行分组和计数 spark 数据帧答案

【问题标题】：Group and count spark dataframe for approxSimilarityJoin对 approxSimilarityJoin 进行分组和计数 spark 数据帧
【发布时间】：2018-03-12 11:48:58
【问题描述】：

如果我们有一个 model.approxSimilarityJoin 的数据框输出

val results =
  model
   .approxSimilarityJoin(vectorizedDf, vectorizedDf, threshold)
   .filter("distCol != 0")
   .filter("distCol < 0.2")
   .select(col("datasetA.title").alias("idA"), col("datasetB.title").alias("idB"), col("distCol"))

上述命令的输出

**idA|idB|distCol**  
A|B|0.125 
B|C|0.125 
A|D|0.125 
D|E|0.125  
F|G|0.125 
X|Y|0.19 
A|M|0.14 
A|N|0.14

我们希望对输出进行分组并计算相似的项目，即在上面的示例中我们有

A, B, C, D, E  
F,G
X,Y

需要的最终输出应该是这样的：

A, 0.125, 5
F, 0.19, 1
A, 0.14, 2

【问题讨论】：

标签： scala apache-spark spark-dataframe sentence-similarity

【解决方案1】：

在 spark-testing-base 中有一个叫做 approx DataFrame 相等性检查的东西

  /**
    * Compares if two [[DataFrame]]s are equal, checks that the schemas are the same.
    * When comparing inexact fields uses tol.
    *
    * @param tol max acceptable tolerance, should be less than 1.
    */
  def assertDataFrameApproximateEquals(
    expected: DataFrame, result: DataFrame, tol: Double) {

    assert(expected.schema, result.schema)

    try {
      expected.rdd.cache
      result.rdd.cache
      assert("Length not Equal", expected.rdd.count, result.rdd.count)

      val expectedIndexValue = zipWithIndex(expected.rdd)
      val resultIndexValue = zipWithIndex(result.rdd)

      val unequalRDD = expectedIndexValue.join(resultIndexValue).
        filter{case (idx, (r1, r2)) =>
          !DataFrameSuiteBase.approxEquals(r1, r2, tol)}

      assertEmpty(unequalRDD.take(maxUnequalRowsToShow))
    } finally {
      expected.rdd.unpersist()
      result.rdd.unpersist()
    }
  }

根据您的需要设计算法。 https://github.com/holdenk/spark-testing-base/wiki/DataFrameSuiteBase

【讨论】：

【解决方案2】：

我们正在尝试寻找类似 Natural Join 的东西。在上面的例子中， A与B相关，B与C相关，即A与C相关。另外我们有 A 与 D 相关，D 与 E 相关，即 A 与 E 相关

最后我们应该得出结论，A 与 B、C、D 和 E 具有相同的相似性即 A 数到 5

【讨论】：