First fixes approximate

src/main/scala/shared/predictions.scala

@@ -249,22 +249,25 @@ package object predictions
   }
 
 
-
-
 //5
 
-def distributed_knn_approximate(preprocessed_ratings : CSCMatrix[Double], k : Int, spark_context: SparkContext, nbPartitions : Int, replication : Int) : Array[Int] = {
-      val new_ratings = new CSCMatrix[Double](ratings.rows, ratings.cols)
-  //Seq[Set[Int]
-      users_partition = partitionUsers (preprocessed_ratings.rows, nbPartitions,replication)
-      val broadcast = sc.broadcast(preprocessed_ratings.toDense)
+def distributed_knn_approximate(preprocessed_ratings : DenseMatrix[Double], k : Int, spark_context: SparkContext, nbPartitions : Int, replication : Int) : DenseMatrix[Double] = {
+      val new_ratings = new CSCMatrix[Double](preprocessed_ratings.rows, preprocessed_ratings.cols)
+
+      val users_partition = partitionUsers (preprocessed_ratings.rows, nbPartitions,replication)
+      val broadcast = sc.broadcast(preprocessed_ratings)
       val approximate_topk = sc.parallelize(users_partition).map(partition_iterator => {
+        
         val ratings = broadcast.value
-        val partition = ratings(partition_iterator,::)
+        val all_users = (0 until ratings.rows).toSeq
+        val exclude_users =all_users.diff(partition_iterator.toSeq)
+        val partition = ratings.delete(exclude_users,Axis._0)
         val similarities = partition * partition.t
-        val partition_index = partition.zipWith(Array[Int](partition.rows))
+        //val partition_index = partition.zipWith(Array[Int](partition.rows))
         //TODO how to integrate indexes in knn
-        val topk = partition_index.map(x => knn(x,k,similarities))
+        val sorted_users =partition_iterator.toArray.sorted
+        val topk = partition_iterator.map(x => (x,knn(sorted_users.indexOf(x),k,similarities)))
+        topk
       }).collect()
 
       //redo knn