Runtime in spark vs not

The following features are different ways to implement a section of anomaly detection code using distance-based algorithms, using scala and spark. Both have the same objective but the second is focused on neighborhoods with the aim of reducing execution time.

def stage1(neighbors: Iterator[Tupla], neighborsNew: Broadcast[Array[Tupla]], k: Int, spark: SparkSession): Iterator[TuplaFase1] = {
        neighbors.toArray.map { neighbor =>
            var distances = neighbor.distance 
            neighborsNew.value.foreach { neighborNew => 
                distances = insert(euclidean(neighbor.values, neighborNew.values, spark), distances, k, spark) 
            }
            Data1(neighbor.id, neighbor.values, IA(distances, spark), distances)
        }.iterator
    }


def stage1(neighborhoods: Iterator[Neighborhood], k: Int, spark: SparkSession): Iterator[TuplaFase1] = {
        neighborhoods.flatMap { neighborhood =>
            neighborhood.neighbors.map { neighbor =>
                var distances =  neighbor.distance
                neighborhood.neighborsNew.foreach { neighborNew =>
                    distances = insert(euclidean(neighbor.values, neighborNew.values, spark), distances, k, spark)
                }
                Data1(neighbor.id, neighbor.values, IA(distances, spark), distances)
            }
        }
    }

In the tests carried out, the results coincide, but the execution times increase considerably. For the first version when the time is 1.294 seconds, in the second with the neighborhood approach it is 5.375 seconds. I don't know why this happens.