无法运行读取元组 RDD 并返回元组 RDD 的 spark map 函数

Question

我需要从另一个配对的 RDD 生成配对的 RDD。基本上，我正在尝试编写一个执行以下操作的地图函数。

RDD[Polygon,HashSet[Point]] => RDD[Polygon,Integer]

这是我写的代码：

迭代 HashSet 并从“Point”对象中累加一个值的 Scala 函数。

def outCountPerCell( jr: Tuple2[Polygon,HashSet[Point]] ) : Tuple2[Polygon,Integer] = {
  val setIter = jr._2.iterator()
  var outageCnt: Int = 0
  while(setIter.hasNext()) {
    outageCnt += setIter.next().getCoordinate().getOrdinate(2).toInt
  }
  return Tuple2(jr._1,Integer.valueOf(outageCnt))
}

在配对的 RDD 上应用该函数，这会引发错误：

scala> val mappedJoinResult = joinResult.map((t: Tuple2[Polygon,HashSet[Point]]) => outCountPerCell(t))
<console>:82: error: type mismatch;
found   : ((com.vividsolutions.jts.geom.Polygon, java.util.HashSet[com.vividsolutions.jts.geom.Point])) => (com.vividsolutions.jts.geom.Polygon, Integer)
required: org.apache.spark.api.java.function.Function[(com.vividsolutions.jts.geom.Polygon, java.util.HashSet[com.vividsolutions.jts.geom.Point]),?]
       val mappedJoinResult = joinResult.map((t: Tuple2[Polygon,HashSet[Point]]) => outCountPerCell(t))

有人可以看看我缺少什么，或者分享任何在 map() 操作中使用自定义函数的示例代码。

Answer 1

这里的问题是 joinResult 是来自 Java API 的 JavaPairRDD。此数据结构的 map 期望 Java 类型 lambdas (Function) 不能（至少是微不足道地）与 Scala lambdas 互换。

因此有两种解决方案：尝试将给定方法转换为 Java Function 以传递给 map，或者按照开发人员的意图简单地使用 Scala RDD：

设置虚拟数据

在这里，我创建了一些备用类，并制作了一个与 OP 结构相似的 Java RDD：

scala> case class Polygon(name: String)
defined class Polygon

scala> case class Point(ordinate: Int)
defined class Point

scala> :pa
// Entering paste mode (ctrl-D to finish)

/* More idiomatic method */
def outCountPerCell( jr: (Polygon,java.util.HashSet[Point])) : (Polygon, Integer) =
{
    val count = jr._2.asScala.map(_.ordinate).sum
    (jr._1, count)
}

// Exiting paste mode, now interpreting.

outCountPerCell: (jr: (Polygon, java.util.HashSet[Point]))(Polygon, Integer)

scala> val hs = new java.util.HashSet[Point]()
hs: java.util.HashSet[Point] = []

scala> hs.add(Point(2))
res13: Boolean = true

scala> hs.add(Point(3))
res14: Boolean = true

scala> val javaRDD = new JavaPairRDD(sc.parallelize(Seq((Polygon("a"), hs))))
javaRDD: org.apache.spark.api.java.JavaPairRDD[Polygon,java.util.HashSet[Point]] = org.apache.spark.api.java.JavaPairRDD@14fc37a

使用 Scala RDD

可以使用.rdd从Java RDD中检索底层Scala RDD：

scala> javaRDD.rdd.map(outCountPerCell).foreach(println)
(Polygon(a),5)

更好的是，使用 mapValues 和 Scala RDD

由于只有元组的第二部分发生了变化，这个问题可以用.mapValues 彻底解决：

scala> javaRDD.rdd.mapValues(_.asScala.map(_.ordinate).sum).foreach(println)
(Polygon(a),5)