Спарк MLlib - предупреждение trainImplicit

Question

Я продолжаю видеть эти предупреждения при использовании trainImplicit:

WARN TaskSetManager: Stage 246 contains a task of very large size (208 KB). 
The maximum recommended task size is 100 KB. 

И тогда размер задача начинает увеличиваться. Я попытался вызвать repartition на входе RDD, но предупреждения одинаковы.

Все эти предупреждения поступают из итераций ALS, из flatMap, а также из совокупности, например, начала этапа, на котором flatMap показывает эти предупреждения (с Spark 1.3.0, но они также показаны в Spark 1.3. 1):

org.apache.spark.rdd.RDD.flatMap(RDD.scala:296) 
org.apache.spark.ml.recommendation.ALS$.org$apache$spark$ml$recommendation$ALS$$computeFactors(ALS.scala:1065) 
org.apache.spark.ml.recommendation.ALS$$anonfun$train$3.apply(ALS.scala:530) 
org.apache.spark.ml.recommendation.ALS$$anonfun$train$3.apply(ALS.scala:527) 
scala.collection.immutable.Range.foreach(Range.scala:141) 
org.apache.spark.ml.recommendation.ALS$.train(ALS.scala:527) 
org.apache.spark.mllib.recommendation.ALS.run(ALS.scala:203) 

и от агрегатный:

org.apache.spark.rdd.RDD.aggregate(RDD.scala:968) 
org.apache.spark.ml.recommendation.ALS$.computeYtY(ALS.scala:1112) 
org.apache.spark.ml.recommendation.ALS$.org$apache$spark$ml$recommendation$ALS$$computeFactors(ALS.scala:1064) 
org.apache.spark.ml.recommendation.ALS$$anonfun$train$3.apply(ALS.scala:538) 
org.apache.spark.ml.recommendation.ALS$$anonfun$train$3.apply(ALS.scala:527) 
scala.collection.immutable.Range.foreach(Range.scala:141) 
org.apache.spark.ml.recommendation.ALS$.train(ALS.scala:527) 
org.apache.spark.mllib.recommendation.ALS.run(ALS.scala:203) 

Answer 1

Аналогичная проблема была описана в Apache Спарк почтовых списков - http://apache-spark-user-list.1001560.n3.nabble.com/Large-Task-Size-td9539.html

I думаю, что вы можете попытаться сыграть с количеством разделов (используя метод repartition()), в зависимости от количества хостов, ОЗУ, процессоров у вас есть.

Попробуйте также изучить все этапы с помощью веб-интерфейса, где вы можете увидеть количество этапов, использование памяти на каждом этапе и местоположение данных.

Или просто не обращайте внимание на эти предупреждения, если все не работает правильно и быстро.

Это уведомление жестко закодировано в Спарк (планировщик/TaskSetManager.scala)

 if (serializedTask.limit > TaskSetManager.TASK_SIZE_TO_WARN_KB * 1024 && 
      !emittedTaskSizeWarning) { 
     emittedTaskSizeWarning = true 
     logWarning(s"Stage ${task.stageId} contains a task of very large size " + 
      s"(${serializedTask.limit/1024} KB). The maximum recommended task size is " + 
      s"${TaskSetManager.TASK_SIZE_TO_WARN_KB} KB.") 
     } 

.

private[spark] object TaskSetManager { 
    // The user will be warned if any stages contain a task that has a serialized size greater than 
    // this. 
    val TASK_SIZE_TO_WARN_KB = 100 
}