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

У меня есть два 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

ИЗМЕНИТЬ:

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

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