Spark中的FPGrowth算法答案

【问题标题】：FPGrowth Algorithm in SparkSpark中的FPGrowth算法
【发布时间】：2017-01-25 19:27:13
【问题描述】：

我正在尝试在 Spark 中运行 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)

我什至尝试使用这里提出的解决方案： SPARK-7483

我也没有运气。有没有人找到解决方案？或者有人知道查看结果或将结果保存到文本文件的方法吗？

任何帮助将不胜感激！

我还找到了该算法的完整源代码 - http://mail-archives.apache.org/mod_mbox/spark-commits/201502.mbox/%3C1cfe817dfdbf47e3bbb657ab343dcf82@git.apache.org%3E

【问题讨论】：

在我想出的最简单的示例数据集中运行时，我也会遇到错误。我得到某种类型转换错误。如果您在自己的方面取得了一些进展，请分享您的发现。谢谢

标签： algorithm scala apache-spark

【解决方案1】：

Kryo 是比 org.apache.spark.serializer.JavaSerializer 更快的序列化程序。一种可能的解决方法是告诉 spark 不要使用 Kryo（至少在修复此错误之前）。您可以修改“spark-defaults.conf”，但 Kryo 对其他 spark 库工作正常。所以最好的方法是修改你的上下文：

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

它现在应该可以工作了。

【讨论】：

【解决方案2】：

我得到了同样的错误：这是因为火花版本。在 Spark 1.5.2 中，这是固定的，但是我使用的是 1.3。我通过执行以下操作进行了修复：

我从使用 spark-shell 切换到 spark-submit，然后更改了 kryoserializer 的配置。这是我的代码：

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

  }
}

【讨论】：

【解决方案3】：

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

【讨论】：