scala - reduceByKey 使用 Scala 对象作为键

Question

我正在将 spark 与 scala 一起使用，并且我有一个充满 tuple2 的 RDD，其中包含一个复杂对象作为键和一个 double 对象。如果对象相同，则目的是添加双倍(频率)。

为此，我将我的对象定义如下:

    case class SimpleCoocurrence(word:String, word_pos:String, cooc:String, cooc_pos:String, distance:Double) extends Ordered[SimpleCoocurrence]{
      def compare(that: SimpleCoocurrence) = {
        if(this.word.equals(that.word)&&this.word_pos.equals(that.word_pos)
           &&this.cooc.equals(that.cooc)&&this.cooc_pos.equals(that.cooc_pos))
          0
        else
          this.toString.compareTo(that.toString)
      }
    }

现在我正在尝试像这样使用 reduceBykey :

val coocRDD = sc.parallelize(coocList)
println(coocRDD.count)
coocRDD.map(tup=>tup).reduceByKey(_+_)
println(coocRDD.count)

但是，结果显示处理reducebykey之前和之后的RDD包含完全相同数量的元素。

如何使用 tuple2[SimpleCoocurrence,Double] 执行 reduceByKey？
实现 Ordered trait 是告诉 Spark 如何比较我的对象的好方法吗？
我应该只使用 tuple2[String,Double] 吗？

谢谢，

Answer 1

reduceByKey不使用 Ordering 但 hashCode和 equals以确定哪些键是相同的。特别是hashPartitioner将按散列对键进行分组，以便具有相同散列码的键落在同一分区上，从而可以在每个分区上进一步减少。

case 类的默认实现是 equals和 hashCode .可能使用的测试数据具有不同的字段值distance:Double使每个实例成为唯一的对象。使用它作为键将导致只有相同的对象被减少为一个。

解决此问题的一种方法是为您的 case class 定义一个 key 。以及对象的添加方法，如下所示:

case class SimpleCoocurrence(word:String, word_pos:String, cooc:String, cooc_pos:String, distance:Double) extends Serializable {
   val key = word + word_pos + cooc + cooc_pos
}
object SimpleCoocurrence {
   val add: (SimpleCoocurrence, SimpleCoocurrence) => SimpleCoocurrence = ???
}

val coocList:List[SimpleCoocurrence] = ???
val coocRDD = sc.parallelize(coocList)
val coocByKey = coocRDD.keyBy(_.key)
val addedCooc = coocByKey.reduceByKey(SimpleCoocurrence.add)

(*) 作为指导示例提供的代码 - 未编译或测试。