Apache Spark 过滤器元素答案

【问题标题】：Apache Spark filter elementsApache Spark 过滤器元素
【发布时间】：2014-10-26 04:00:51
【问题描述】：

我有两个 RDD：points 和 pointsWithinEps。它们的内容如下图所示：

向量代表x, y坐标。 pointsWithinEps 代表两点和它们之间的距离。我想循环所有points，并且对于每个点，仅将pointsWithinEps 中的元素过滤为x（第一个）坐标。所以对于第一点，它将给出来自pointsWithinEps 的[0] 和[1] 向量。我有以下代码：

for (i <- 0 until points.count.toInt) {
  val p = points.take(i + 1).drop(i)
  val currentPointNeighbours = pointsWithinEps.filter {
    case ((x, y), distance) =>
      x == p
  }
  currentPointNeighbours.foreach(println)
  println("----")
}

它不能正常工作。代码有什么问题？

【问题讨论】：

标签： scala foreach filter apache-spark

【解决方案1】：

如果您将 RDD 转换为 K-V RDD，然后在 key 上加入，您可以有效地做到这一点。对于点，关键是点本身，对于距离，关键是第一个点

  import org.apache.spark.SparkContext._

  type Point = DenseVector
  type Distance = ((Point, Point), Double)

  val points: RDD[Point] = ???
  val pointsWithinEps: RDD[Distance] = ???

  // Prepare Tuple2 RDD to enable spark tuple functions
  val pointsToKV: RDD[(Point, Unit)] = points.map(p => p -> ())
  val distance: RDD[(Point, Distance)] = pointsWithinEps.map(distance => distance._1._1 -> distance)

  // Join points with distance
  val filtered: RDD[Distance] = pointsToKV.join(distance) map (_._2._2)

【讨论】：