0

拜托,你能帮帮我吗?我有一个 80 个 CSV 文件数据集和一个由一个主设备和 4 个从设备组成的集群。我想读取数据帧中的 CSV 文件并将其并行化到四个从站上。之后,我想用 group by 过滤数据框。在我的 spark 查询中,结果包含按(“code_ccam”、“dossier”)分组的“code_ccam”和“dossier”列。我想使用 FP-Growth 算法来检测由“文件夹”重复的“code_ccam”序列。但是当我使用 FPGrowth.fit() 命令时,出现以下错误:

"error: type mismatch;
found : org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
required: org.apache.spark.sql.Dataset[_]"

这是我的火花命令:

val df = spark.read.option("header", "true").csv("file:///home/ia/Projet-Spark-ace/Donnees/Fichiers CSV/*.csv")
import org.apache.spark.sql.functions.{concat, lit}
val df2 = df.withColumn("dossier", concat(col("num_immatriculation"), lit(""), col("date_acte"), lit(""), col("rang_naissance"), lit(""), col("date_naissance")))
val df3 = df2.drop("num_immatriculation").drop("date_acte").drop("rang_naissance").drop("date_naissance")
val df4 = df3.select("dossier","code_ccam").groupBy("dossier","code_ccam").count()
val transactions = df4.agg(collect_list("code_ccam").alias("codes_ccam")).rdd.map(x => x)
import org.apache.spark.ml.fpm.FPGrowth
val fpgrowth = new FPGrowth().setItemsCol("code_ccam").setMinSupport(0.5).setMinConfidence(0.6)
val model = fpgrowth.fit(transactions)
4

1 回答 1

0

非常感谢你。有效。我用 collect_set 替换了 collect_list。

于 2019-02-20T19:53:00.833 回答