title: Generating recommendation model for a large dataset using spark
body: <p>I'm trying out to generate a simple ALS model using the spark documentation <a href="https://spark.apache.org/docs/latest/mllib-collaborative-filtering.html" rel="nofollow noreferrer">here</a>.
  My first file(ratings.csv) has 20million UserID,MovID,Rat and can be downloaded <a href="http://ml-20m.zip" rel="nofollow noreferrer">here</a></p>
  
  <p>So I have the testing data which is a subset of ratings.csv. That test dataset can be downloaded <a href="https://drive.google.com/open?id=0B_z7iObE9G7lTGtROG1LRmFONlk" rel="nofollow noreferrer">here</a>:
  The test file has just the UserID, Movie ID column.</p>
  
  <p>So to create training data we will have to filter ratings.csv.
  The following code is working fine for a smaller case of 100,000 UserID,MovID rating. I am not able to generate the model for the big case. 
  Please help with a pointer. </p>
  
  <pre><code>/**
    * Created by echoesofconc on 3/8/17.
    */
  import org.apache.spark.SparkConf
  import org.apache.spark.SparkContext
  import org.apache.spark.rdd.RDD
  import org.apache.spark.mllib.recommendation.ALS
  import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
  import org.apache.spark.mllib.recommendation.Rating
  import java.io._
  import scala.collection.mutable.ListBuffer
  
  
  object Prateek_Agrawal_task1 {
  
    def dropheader(data: RDD[String]): RDD[String] = {
      data.mapPartitionsWithIndex((idx, lines) =&gt; {
        if (idx == 0) {
          lines.drop(1)
        }
        lines
      })
    }
  
    def create_training(ratings_split: RDD[Array[String]], ratings_testing: Array[Array[String]]) = {
      ratings_split.filter(x =&gt; {
        ratings_testing.exists(y =&gt;
          (x(0) == y(0) &amp;&amp; x(1) == y(1))
        ) == false
      })
    }
    def create_testing(ratings_split: RDD[Array[String]], ratings_testing: Array[Array[String]]) = {
      ratings_split.filter(x =&gt; {
        ratings_testing.exists(y =&gt;
          (x(0) == y(0) &amp;&amp; x(1) == y(1))
        ) == true
      })
    }
    def create_model(ratings_train:RDD[Array[String]],rank:Int,numIterations:Int ):org.apache.spark.mllib.recommendation.MatrixFactorizationModel={
      val ratings = ratings_train.map(_ match { case Array(user,item,rate,temp) =&gt;
        Rating(user.toInt, item.toInt, rate.toDouble)
      })
      val model = ALS.train(ratings, rank, numIterations, 0.01)
      return model
    }
  
    def print_results(final_predictions_adjusted:RDD[((Int, Int), Double)])={
      val rating_range=final_predictions_adjusted.map(x=&gt;(x._2.toInt,1)).reduceByKey(_+_).sortByKey()
      val rating_range_till_4=rating_range.map{x=&gt;
        var temp=x
        if (x._1==5){temp=(4,x._2)}
        temp
      }.reduceByKey(_+_)
      rating_range_till_4.sortByKey().foreach { x =&gt;
        if(x._1==0)
          printf("&gt;=0 and &lt;1: " + x._2+"\n")
        if(x._1==1)
          printf("&gt;=1 and &lt;2: " + x._2+"\n")
        if(x._1==2)
          printf("&gt;=2 and &lt;3: " + x._2+"\n")
        if(x._1==3)
          printf("&gt;=3 and &lt;4: " + x._2+"\n")
        if(x._1==4)
          printf("&gt;=4 " + x._2+"\n")
        if(x._1==5)
          printf("=5 " + x._2+"\n")
      }
    }
    case class User_mov_rat(UserID: Int, MovieID:Int, Pred_rating: Double)
  
    def print_outputfile(final_predictions_adjusted:RDD[((Int, Int), Double)])={
      val writer = new FileWriter(new File("./output.txt" ))
      writer.write("UserID,MovieID,Pred_rating\n")
      final_predictions_adjusted.collect().foreach(x=&gt;{writer.write(x._1._1+","+x._1._2+","+x._2+"\n")})
      writer.close()
    }
  
    def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setAppName("Prateek_Agrawal_task1").setMaster("local[2]")
      val sc = new SparkContext(conf)
  
  
     val file = "/Users/echoesofconc/Documents/USC_courses/INF553/ml-20m/ratings.csv"
     val test = "/Users/echoesofconc/Documents/USC_courses/INF553/Prateek_Agrawal_hw3/testing_20m.csv"
  
      val data = sc.textFile(file, 2).cache()
      val data_test = sc.textFile(test, 2).cache()
  
  //    Drop Header
      val data_wo_header=dropheader(data).persist()
      val data_test_wo_header=dropheader(data_test).persist()
  //    Create Training and testing data of the format (User ID, MovID, Rating, Time)
  
      val ratings_split = data_wo_header.map(line =&gt; line.split(",")).persist()
      data_wo_header.unpersist()
      data.unpersist()
      val ratings_testing = data_test_wo_header.map(line =&gt; line.split(",")).collect()
      data_test_wo_header.unpersist()
      data_test.unpersist()
  
      val ratings_train = create_training(ratings_split, ratings_testing).persist()
      val ratings_test=create_testing(ratings_split, ratings_testing)
      ratings_split.unpersist()
      ratings_test.unpersist()
  
  //    Create the model using rating_train the training data
      val rank = 1
      val numIterations = 10
  
      val model=create_model(ratings_train,rank,numIterations)
      ratings_train.unpersist()
  
  //    Average user,Rating from training this is for cases which are there in test but not rated by any user in training
      val user_avgrat=ratings_test.map(_ match { case Array(user, mov, rate, temp) =&gt;(user.toInt, (rate.toDouble,1.0))}).reduceByKey((x,y)=&gt;(x._1 + y._1, x._2 + y._2)).mapValues{ case (sum, count) =&gt; (1.0 * sum) / count }
  
  //    Predict user_mov ratings
      val user_mov = data_test_wo_header.map(_.split(',') match { case Array(user, mov) =&gt;
        (user.toInt,mov.toInt)
      })
  
      val predictions =
      model.predict(user_mov).map { case Rating(user, mov, rate) =&gt;
        ((user, mov), rate)
      }
  //    Combine Predictions and unpredicted user,Movies due to them being individual. Going forward we need to improve the accuracy for these predictions
      val user_mov_rat=user_mov.map(x=&gt;(x,0.0))
      val predictions_unpredicted_combined= predictions.union(user_mov_rat).reduceByKey(_+_).map(x=&gt;(x._1._1,(x._1._2,x._2)))
  //    Combine average rating and predictions+unpredicted values
      val avg_rating_predictions_unpredicted_combined=predictions_unpredicted_combined.join(user_avgrat)
  //    Generate final predictions RDD
      val final_predictions=avg_rating_predictions_unpredicted_combined.map{x=&gt;
        var temp=((x._1,x._2._1._1),x._2._2)
        if(x._2._1._2==0.0){temp=((x._1,x._2._1._1),x._2._2)}
        if(x._2._1._2!=0.0){temp=((x._1,x._2._1._1),x._2._1._2)}
        temp
      }
  //    Adjust for ratings above 5.0 and below 0.0
      val final_predictions_adjusted=final_predictions.map{x=&gt;
        var temp=x
        if (x._2&gt;5.0){temp=(x._1,5.0)}
        if (x._2&lt;0.0){temp=(x._1,0.0)}
        temp
      }
      val ratesAndPreds = ratings_test.map(_ match { case Array(user, mov, rate, temp) =&gt; ((user.toInt,mov.toInt),rate.toDouble)}).join(final_predictions_adjusted)
  
      val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =&gt;
        val err = (r1 - r2)
        err * err
      }.mean()
      val RMSE=math.sqrt(MSE)
  //    Print output.txt
      print_outputfile(final_predictions_adjusted)
  //    Print the predictionresults
      print_results(final_predictions_adjusted.sortByKey())
      print(RMSE+"\n")
    }
  }
  </code></pre>
  
  <p>In case someone thinks I should be doing a regex match I have tried that approach. BUt that dosen't seem to be a bottleneck. </p>
  
  <p>I only need to complete the create model part on which I am stuck for the big dataset. Can somebody help.</p>
  
  <p>EDIT:
  Another approach I tried which is much faster by using broadcast variables. But it's been running for 12 hrs with no signs of progress. On spark UI somehow the whole of the RDD(ratings.csv ~500MB) is not cached. Only around 64MB with 2.5 Million lines is being processed initially. I am using <code>--executor-memory -8g</code>. I have modified the <code>create_training create_testing</code> functions:</p>
  
  <pre><code>/**
    * Created by echoesofconc on 3/8/17.
    */
  import org.apache.spark.SparkConf
  import org.apache.spark.SparkContext
  import org.apache.spark.rdd.RDD
  import org.apache.spark.mllib.recommendation.ALS
  import org.apache.spark.mllib.recommendation.Rating
  import java.io._
  
  
  
  object Prateek_Agrawal_task2 {
  
    def dropheader(data: RDD[String]): RDD[String] = {
      data.mapPartitionsWithIndex((idx, lines) =&gt; {
        if (idx == 0) {
          lines.drop(1)
        }
        lines
      })
    }
    def create_training(data_wo_header: RDD[String], data_test_wo_header: RDD[String],sc:SparkContext): RDD[String] = {
  
      val rdd2array = sc.broadcast(data_test_wo_header.collect())
      val training_set = data_wo_header.filter{
        case(x) =&gt; rdd2array.value.filter(y =&gt; x.indexOf(y.toString())==0).length == 0
      }
      return training_set
    }
    def create_test(data_wo_header: RDD[String], data_test_wo_header: RDD[String],sc:SparkContext): RDD[String] = {
  
      val rdd2array = sc.broadcast(data_test_wo_header.collect())
      val training_set = data_wo_header.filter{
        case(x) =&gt; rdd2array.value.filter(y =&gt; x.indexOf(y.toString())==0).length != 0
      }
      return training_set
    }
  
    def create_model(ratings_train:RDD[String],rank:Int,numIterations:Int ):org.apache.spark.mllib.recommendation.MatrixFactorizationModel={
      val ratings = ratings_train.map(_.split(',') match { case Array(user, item, rate, timestamp) =&gt;
        Rating(user.toInt, item.toInt, rate.toDouble)
      })
      val model = ALS.train(ratings, rank, numIterations, 0.01)
      return model
    }
  
    def print_results(final_predictions_adjusted:RDD[((Int, Int), Double)])={
      val rating_range=final_predictions_adjusted.map(x=&gt;(x._2.toInt,1)).reduceByKey(_+_).sortByKey()
      val rating_range_till_4=rating_range.map{x=&gt;
        var temp=x
        if (x._1==5){temp=(4,x._2)}
        temp
      }.reduceByKey(_+_)
      rating_range_till_4.sortByKey().foreach { x =&gt;
        if(x._1==0)
          printf("&gt;=0 and &lt;1: " + x._2+"\n")
        if(x._1==1)
          printf("&gt;=1 and &lt;2: " + x._2+"\n")
        if(x._1==2)
          printf("&gt;=2 and &lt;3: " + x._2+"\n")
        if(x._1==3)
          printf("&gt;=3 and &lt;4: " + x._2+"\n")
        if(x._1==4)
          printf("&gt;=4 " + x._2+"\n")
        if(x._1==5)
          printf("=5 " + x._2+"\n")
      }
    }
    case class User_mov_rat(UserID: Int, MovieID:Int, Pred_rating: Double)
  
    def print_outputfile(final_predictions_adjusted:RDD[((Int, Int), Double)])={
      val writer = new FileWriter(new File("./output.txt" ))
      writer.write("UserID,MovieID,Pred_rating\n")
      final_predictions_adjusted.collect().foreach(x=&gt;{writer.write(x._1._1+","+x._1._2+","+x._2+"\n")})
      writer.close()
    }
  
    def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setAppName("Prateek_Agrawal_task1").setMaster("local[2]")
      val sc = new SparkContext(conf)
  
      val file = "/Users/echoesofconc/Documents/USC_courses/INF553/ml-latest-small/ratings.csv"
      val test = "/Users/echoesofconc/Documents/USC_courses/INF553/Prateek_Agrawal_hw3/testing_small.csv"
  
  //    val file = "/Users/echoesofconc/Documents/USC_courses/INF553/ml-20m/ratings.csv"
  //    val test = "/Users/echoesofconc/Documents/USC_courses/INF553/Prateek_Agrawal_hw3/testing_20m.csv"
  
      val data = sc.textFile(file, 2).persist()
      val data_test = sc.textFile(test, 2).persist()
  
      //    Drop Header
      val data_wo_header=dropheader(data)
      val data_test_wo_header=dropheader(data_test)
      //    Create Traing and testing data of the format (User ID, MovID, Rating, Time)
  
      val ratings_train=create_training(data_wo_header,data_test_wo_header,sc).persist()
      val ratings_test=create_test(data_wo_header,data_test_wo_header,sc)
  //    val ratings_test=create_test(data_wo_header,data_test_wo_header,sc)
  
  
  //    data_test_wo_header.unpersist()
  //    data_test.unpersist()
  ////    data.unpersist()
  ////    data_test.unpersist()
      //    Create the model using rating_train the training data
      val rank = 1
      val numIterations = 10
      val model=create_model(ratings_train,rank,numIterations)
  
  
  
  
      //    ratings_train.unpersist()
      //    model.save(sc, "target/tmp/myCollaborativeFilter")
      //    val Model = MatrixFactorizationModel.load(sc, "/Users/echoesofconc/myCollaborativeFilter")
  
      //    Average user,Rating from training
      val user_avgrat=ratings_test.map(_.split(",") match { case Array(user, mov, rate, temp) =&gt;(user.toInt, (rate.toDouble,1.0))}).reduceByKey((x,y)=&gt;(x._1 + y._1, x._2 + y._2)).mapValues{ case (sum, count) =&gt; (1.0 * sum) / count }
      //data
      //    Predict user_mov ratings
      val user_mov = data_test_wo_header.map(_.split(',') match { case Array(user, mov) =&gt;
        (user.toInt,mov.toInt)
      })
  
      val predictions =
        model.predict(user_mov).map { case Rating(user, mov, rate) =&gt;
          ((user, mov), rate)
        }
      //    Combine Predictions and unpredicted user,Movies due to them being individual. Going forward we need to improve the accuracy for these predictions
      val user_mov_rat=user_mov.map(x=&gt;(x,0.0))
      val predictions_unpredicted_combined= predictions.union(user_mov_rat).reduceByKey(_+_).map(x=&gt;(x._1._1,(x._1._2,x._2)))
      //    Combine average rating and predictions+unpredicted values
      val avg_rating_predictions_unpredicted_combined=predictions_unpredicted_combined.join(user_avgrat)
      //    Generate final predictions RDD
      val final_predictions=avg_rating_predictions_unpredicted_combined.map{x=&gt;
        var temp=((x._1,x._2._1._1),x._2._2)
        if(x._2._1._2==0.0){temp=((x._1,x._2._1._1),x._2._2)}
        if(x._2._1._2!=0.0){temp=((x._1,x._2._1._1),x._2._1._2)}
        temp
      }
      //    Adjust for ratings above 5.0 and below 0.0
      val final_predictions_adjusted=final_predictions.map{x=&gt;
        var temp=x
        if (x._2&gt;5.0){temp=(x._1,5.0)}
        if (x._2&lt;0.0){temp=(x._1,0.0)}
        temp
      }
      val ratesAndPreds = ratings_test.map(_.split(",") match { case Array(user, mov, rate, temp) =&gt; ((user.toInt,mov.toInt),rate.toDouble)}).join(final_predictions_adjusted)
  
      val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =&gt;
        val err = (r1 - r2)
        err * err
      }.mean()
      val RMSE=math.sqrt(MSE)
      //    Print output.txt
      print_outputfile(final_predictions_adjusted)
      //    Print the predictionresults
      print_results(final_predictions_adjusted.sortByKey())
      print(RMSE+"\n")
    }
  }
  </code></pre>