星火中的FPGrowth算法

Question

我试图在星火中运行一个FPGrowth算法的例子，但是我遇到了一个错误。这是我的密码：

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)}

代码一直工作到我得到错误的最后一行：

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)

我甚至尝试使用这里提出的解决方案：火花-7483

这件事我也没有任何运气。有人找到解决办法了吗？或者有人知道一种只查看结果或将结果保存到文本文件的方法吗？

任何帮助都将不胜感激！

我还找到了这个算法的完整源代码- mbox/spark-commits/201502.mbox/%3C1cfe817dfdbf47e3bbb657ab343dcf82@git.apache.org%3E。

Answer 1

我也犯了同样的错误:这是因为火花版本。在Spark1.5.2中，这是固定的，但是我使用的是1.3。我通过做以下几件事来修正：

1. 我从使用火花壳切换到火花提交，然后更改了kryoserializer的配置。这是我的代码： 导入org.apache.spark.{SparkConf，SparkContext}导入org.apache.spark.rdd.RDD org.apache.spark.mllib.fpm.FPGrowth scala.collection.mutable.ArrayBuffer org.apache.spark.mllib.fpm.FPGrowth scala.collection.mutable.ArrayBuffer org.apache.spark.mllib.fpm.FPGrowth scala.collection.mutable.ArrayBuffer scala.collection.mutable.ListBuffer scala.collection.mutable.ListBuffer对象{ def (args: ArrayString) { val =新SparkConf().setAppName(“火花FPGrowth") .registerKryoClasses(数组，classOf[ArrayBufferString]，( classOf[ListBufferString]) ) val sc =新SparkContext(conf) val data = sc.textFile("") val事务: RDD[ArrayString] = data.map(s => s.trim.split(‘)) val fpg =新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是一个比org.apache.spark.serializer.JavaSerializer更快的序列化程序。一个可能的解决办法是告诉possible不要使用Kryo (至少在这个bug被修复之前)。您可以修改“火花-defaults.conf”，但是Kryo在其他火花库中工作得很好。因此，最好的方法是使用以下内容修改上下文：

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

然后尝试再次运行MLLIb代码：

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

现在应该管用了。

Answer 3

在cmd或星火-defaults.conf-conf spark.kryo.classesToRegister=scala.collection.mutable.ArrayBuffer，scala.collection.mutable.ListBuffer中设置下面的配置