0

我有两个 RDD：points和pointsWithinEps. 中的每个点points代表x, y坐标。pointsWithinEps表示两点和它们之间的距离：((x, y), distance). 我想循环所有点，并且为每个点过滤仅位于pointsWithinEpsas x（第一）坐标中的元素。所以我做以下事情：

    points.foreach(p =>
      val distances = pointsWithinEps.filter{
        case((x, y), distance) => x == p
      }
      if (distances.count() > 3) {
//        do some other actions
      }
    )

但是这种语法是无效的。据我了解，不允许在 Spark foreach 中创建变量。我应该做这样的事情吗？

for (i <- 0 to points.count().toInt) {
  val p = points.take(i + 1).drop(i) // take the point
  val distances = pointsWithinEps.filter{
    case((x, y), distance) => x == p
  }
  if (distances.count() > 3) {
    //        do some other actions
  }
}

或者有更好的方法来做到这一点？完整的代码托管在这里：https ://github.com/timasjov/spark-learning/blob/master/src/DBSCAN.scala

编辑：

points.foreach({ p =>
  val pointNeighbours = pointsWithinEps.filter {
    case ((x, y), distance) => x == p
  }
  println(pointNeighbours)
})

现在我有以下代码，但它抛出了 NullPointerException (pointsWithinEps)。如何解决它为什么pointsWithinEps是 null （在 foreach 之前有元素）？

4

2 回答 2

2

为了收集从给定坐标开始的所有距离点，一种简单的分布式方法是按该坐标键控点x并按该键对它们进行分组，如下所示：

val pointsWithinEpsByX = pointsWithinEps.map{case ((x,y),distance) => (x,((x,y),distance))}
val xCoordinatesWithDistance = pointsWithinEpsByX.groupByKey

然后将点的 RDD 与之前转换的结果左连接：

val pointsWithCoordinatesWithDistance = points.leftOuterJoin(xCoordinatesWithDistance)
于 2014-10-26T20:47:45.867 回答
0

声明变量意味着你有一个块，而不仅仅是一个表达式，所以你需要使用大括号{}，例如

point.foreach({p => ... })
于 2014-10-26T19:50:32.143 回答