Выполнение кода внутри Spark foreach

Question

У меня есть два RDD: points и pointsWithinEps. Каждая точка в points представляет координату x, y. pointsWithinEps представляют две точки и расстояние между ними: ((x, y), distance). Я хочу, чтобы петля все точки и для каждого точечного фильтра только те элементы, которые находятся в pointsWithinEps как 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

EDIT:

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

Сейчас я следующий код, но он бросает NullPointerException (pointsWithinEps). Как это можно исправить, почему pointsWithinEps имеет значение null (перед тем, как в нем есть элементы)?

Answer 1

Для того, чтобы собрать все точки расстояния, которые начинаются на заданную координату, простым раздельным способом сделать это было бы ввести точки по этой координате x и сгруппировать их по этой клавише, например:

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

Тогда левая присоединиться к RDD точек с результатом предыдущего преобразования:

val pointsWithCoordinatesWithDistance = points.leftOuterJoin(xCoordinatesWithDistance) 

Answer 2

Объявление переменных означает, что у вас есть блок, а не только выражение, поэтому вам нужно использовать фигурные скобки {}, например.

point.foreach({p => ... })