Algorithme FPGrowth dans Spark

Question

J'essaye d'exécuter un exemple de l'algorithme FPGrowth dans Spark, cependant, je rencontre une erreur. Ceci est mon code:

import org.apache.spark.rdd.RDD 
import org.apache.spark.mllib.fpm.{FPGrowth, FPGrowthModel} 

val transactions: RDD[Array[String]] = sc.textFile("path/transations.txt").map(_.split(" ")).cache() 

val fpg = new FPGrowth().setMinSupport(0.2).setNumPartitions(10) 

val model = fpg.run(transactions) 

model.freqItemsets.collect().foreach { itemset => println(itemset.items.mkString("[", ",", "]") + ", " + itemset.freq)} 

Le code fonctionne jusqu'à la dernière ligne où je reçois l'erreur:

WARN TaskSetManager: Lost task 0.0 in stage 4.0 (TID 16, ip-10-0-0-###.us-west-1.compute.internal): 
com.esotericsoftware.kryo.KryoException: java.lang.IllegalArgumentException: Can not set 
final scala.collection.mutable.ListBuffer field org.apache.spark.mllib.fpm.FPTree$Summary.nodes to scala.collection.mutable.ArrayBuffer 
Serialization trace: 
nodes (org.apache.spark.mllib.fpm.FPTree$Summary) 

J'ai même essayé d'utiliser la solution qui a été proposée ici: SPARK-7483

Je n'ai pas eu de chance avec ça non plus. Est-ce que quelqu'un a trouvé une solution à cela? Ou est-ce que quelqu'un connaît un moyen de voir les résultats ou de les enregistrer dans un fichier texte?

Toute aide serait grandement appréciée!

Je trouve aussi le code source complet pour cet algorithme - http://mail-archives.apache.org/mod_mbox/spark-commits/201502.mbox/%3C1cfe817dfdbf47e3bbb657ab343dcf82@git.apache.org%3E

Answer 1

J'ai la même erreur: C'est à cause de la version étincelle. Dans Spark 1.5.2, ceci est corrigé, mais j'utilisais 1.3. Je fixe en procédant comme suit:

1. Je suis passé de l'aide-shell étincelle pour déclencher soumettre et a changé la configuration pour kryoserializer. Voici mon code:

   import org.apache.spark.{SparkConf, SparkContext} 
   import org.apache.spark.rdd.RDD 
   import org.apache.spark.mllib.fpm.FPGrowth 
   import scala.collection.mutable.ArrayBuffer 
   import scala.collection.mutable.ListBuffer 
   
   object fpgrowth { 
       def main(args: Array[String]) { 
       val conf = new SparkConf().setAppName("Spark FPGrowth") 
        .registerKryoClasses(
        Array(classOf[ArrayBuffer[String]], classOf[ListBuffer[String]]) 
       ) 
   
       val sc = new SparkContext(conf) 
   
       val data = sc.textFile("<path to file.txt>") 
   
       val transactions: RDD[Array[String]] = data.map(s => s.trim.split(' ')) 
   
       val fpg = new FPGrowth() 
        .setMinSupport(0.2) 
        .setNumPartitions(10) 
       val model = fpg.run(transactions) 
   
       model.freqItemsets.collect().foreach { itemset => 
        println(itemset.items.mkString("[", ",", "]") + ", " + itemset.freq) 
       } 
   
       } 
   } 
   

Answer 2

Kryo est un sérialiseur plus rapide que org.apache.spark.serializer.JavaSerializer. Une solution de contournement possible consiste à dire à spark de ne pas utiliser Kryo (au moins jusqu'à ce que ce bogue soit corrigé). Vous pouvez modifier le fichier "spark-defaults.conf", mais Kryo fonctionne correctement pour les autres bibliothèques d'étincelles. Donc, le mieux est de modifier votre contexte avec:

val conf = (new org.apache.spark.SparkConf() 
      .setAppName("APP_NAME") 
      .set("spark.serializer", "org.apache.spark.serializer.JavaSerializer") 

et essayez d'exécuter à nouveau le code MLlib:

model.freqItemsets.collect().foreach { itemset => println(itemset.items.mkString("[", ",", "]") + ", " + itemset.freq)} 

Il devrait maintenant fonctionner.

Answer 3

set config ci-dessous dans cmd ou des étincelles defaults.conf --conf spark.kryo.classesToRegister = scala.collection.mutable.ArrayBuffer, scala.collection.mutable. ListBuffer