reduceSyKey使用Scala对象作为键

发布时间：2020-12-16 18:16:44 所属栏目：安全来源：网络整理

导读：我正在使用带 scala的spark,我有一个充满了tuple2的RDD,其中包含一个复杂的对象作为键和一个double.目标是在对象相同时添加双精度(频率). 为此,我已将我的对象定义如下： case class SimpleCoocurrence(word:String,word_pos:String,cooc:String,cooc_pos:St

我正在使用带 scala的spark,我有一个充满了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]吗？

谢谢,

解决方法

reduceByKey不使用Ordering而是使用hashCode和equals来确定哪些键是相同的.特别是,hashPartitioner将通过散列对密钥进行分组,使得具有相同hashCode的密钥落在同一分区上,以便在每个分区上进一步减少.

case类有一个equals和hashCode的默认实现.可能使用的测试数据具有不同的场距值：双重使每个实例成为唯一的对象.将其用作密钥将导致仅将相同的对象缩减为一个.

解决此问题的一种方法是为您的案例类定义一个键,并为该对象定义一个添加方法,如下所示：

case class SimpleCoocurrence(word: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)

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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!