<br> I try to build a model for movie lens rating data with Spark ALS. On Windows host, I use Spark 2.3.1. Data has just and 100.000 rows and three columns; userid, movieid, and rating. My machine has Intel i7 and 32 GB memory. I have increased executor memory to 10 G. I get java.lang.StackOverflowErrorMy error. My codes are below: object ErkansALS {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf()
.setMaster("local[4]")
.setAppName("SparkALS")
.setExecutorEnv("spark.driver.memory","8g")
.setExecutorEnv("spark.executor.memory","10g")
.setExecutorEnv("spark.sql.broadcastTimeout","1200")
val spark = SparkSession.builder()
.config(sparkConf)
.getOrCreate()
val movieRatings = spark.read.format("csv")
.option("header","true")
.option("inferSchema","true")
.load("ratings.csv")
.drop("timestamp")
val Array(training, test) = movieRatings.randomSplit(Array(0.8, 0.2),seed = 142)
training.cache()
val alsObject = new ALS()
.setUserCol("userId")
.setItemCol("movieId")
.setRatingCol("rating")
.setColdStartStrategy("drop")
.setNonnegative(true)
val paramGridObject = new ParamGridBuilder()
.addGrid(alsObject.rank, Array(12,14))
.addGrid(alsObject.maxIter, Array(18,20))
.addGrid(alsObject.regParam, Array(.17,.19))
.build()
val evaluator = new RegressionEvaluator()
.setMetricName("rmse")
.setLabelCol("rating")
.setPredictionCol("prediction")
val tvs = new TrainValidationSplit()
.setEstimator(alsObject)
.setEstimatorParamMaps(paramGridObject)
.setEvaluator(evaluator)
val model = tvs.fit(training)
val bestModel = model.bestModel
val predictions = bestModel.transform(test)
val rmse = evaluator.evaluate(predictions)
predictions.show()
println("RMSE = ", rmse)
println("Best Model")
}
}
Errors are attached. But when I try without TrainValidationSplit it works: package spark.ml.recommendation.als import org.apache.spark.ml.evaluation.RegressionEvaluator import org.apache.spark.ml.recommendation.ALS import org.apache.spark.ml.tuning.{TrainValidationSplit, ParamGridBuilder} import org.apache.spark.sql.{SparkSession} import org.apache.spark.{SparkConf, SparkContext} object ErkansALS { def main(args: Array[String]): Unit = { /* val sparkConf = new SparkConf() .setExecutorEnv("spark.driver.memory","4g") .setExecutorEnv("spark.executor.memory","8g") .setExecutorEnv("spark.sql.broadcastTimeout","1200") .setExecutorEnv("spark.eventLog.enabled","false")*/val spark = SparkSession.builder() .master("local[*]") .appName("SparkALS") .getOrCreate() val movieRatings = spark.read.format("csv") .option("header","true") .option("inferSchema","true") .load("C:\\Users\\toshiba\\SkyDrive\\veribilimi.co\\Datasets\\ml-latest-small\\ratings.csv") .drop("timestamp") // .sample(0.1,142)movieRatings.show() println(movieRatings.count()) // 100.004 adet rating var. // Create training and test setval Array(training, test) = movieRatings.randomSplit(Array(0.8, 0.2),seed = 142) training.cache() // Create ALS modelval alsObject = new ALS() .setUserCol("userId") .setItemCol("movieId") .setRatingCol("rating") .setColdStartStrategy("drop") .setNonnegative(true) /* // Tune model using ParamGridBuilder val paramGridObject = new ParamGridBuilder() .addGrid(alsObject.rank, Array(14)) .addGrid(alsObject.maxIter, Array(20)) .addGrid(alsObject.regParam, Array(.19)) .build()*/ // Define evaluator as RMSEval evaluator = new RegressionEvaluator() .setMetricName("rmse") .setLabelCol("rating") .setPredictionCol("prediction") /* // Build cross validation using TrainValidationSplit val tvs = new TrainValidationSplit() .setEstimator(alsObject) .setEstimatorParamMaps(paramGridObject) .setEvaluator(evaluator)*/ // Fit ALS model to training setval model = alsObject.fit(training) /* // Take best model val bestModel = model.bestModel*/ // Generate predictions and evaluate RMSEval predictions = model.transform(test) val rmse = evaluator.evaluate(predictions) predictions.show() // Print evaluation metrics and model parametersprintln("RMSE = ", rmse) } }
... View more
