In Apache Spark (Scala shell), I am attempting:
val model = ALS.trainImplicit(training, rank, numIter)
where training is a million-row file partitioned into 100 partitions, rank=20, and numIter=20.
I get a string of messages of the form:
WARN scheduler.TaskSetManager: Stage 2175 contains a task of very large size (101 KB). The maximum recommended task size is 100 KB.
How do I go about debugging this? I've heard broadcast variables are useful in reducing task size, but in this case there's no large variable other than the RDD itself, and it's been partitioned into many chunks already.
The entire code is below (run from Spark shell):
Code
import java.io.File
import scala.io.Source
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.rdd._
import org.apache.spark.mllib.recommendation.{ALS, Rating, MatrixFactorizationModel}
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
// Input file
// This file has about a million rows in the format:
// user::item::rating
val trainfile = "training_rows.txt"
// Function defs
def hash32(x: String) : Int = {
return x.hashCode() & 0x7fffffff
}
def readfile(datafile: String) : RDD[Rating] = {
val numPartitions = 100
sc.textFile(datafile).map { line =>
val fields = line.split("::")
Rating(hash32(fields(0)), hash32(fields(1)), fields(2).toDouble)
}
.repartition(numPartitions).cache()
}
// Main code
val training = readfile(trainfile)
val numTraining = training.count()
println("Training rows: " + numTraining)
val rank = 20
val numIter = 20
val model = ALS.trainImplicit(training, rank, numIter)
this
. Could you post your entire code and we can see if this might be the case? – Speller