pyspark FPGrowth不適用於RDD

Question

我想在Spark中的某些數據上使用FPGrowth函數。我測試的例子在這裏，沒有任何問題： https://spark.apache.org/docs/latest/mllib-frequent-pattern-mining.html

然而，我的數據集從蜂巢

data = hiveContext.sql('select transactionid, itemid from transactions') 
model = FPGrowth.train(data, minSupport=0.1, numPartitions=100) 

這種失敗，法來不存在：

py4j.protocol.Py4JError: An error occurred while calling o764.trainFPGrowthModel. Trace: 
py4j.Py4JException: Method trainFPGrowthModel([class org.apache.spark.sql.DataFrame, class java.lang.Double, class java.lang.Integer]) does not exist 

所以，我把它轉換到RDD：

data=data.rdd 

現在我開始獲取一些s trange pickle序列化錯誤。

net.razorvine.pickle.PickleException: expected zero arguments for construction of ClassDict (for pyspark.sql.types._create_row) 

然後我開始看類型。在該示例中，數據通過平面地圖運行。這將返回與RDD不同的類型。通過flatmap返回

RDD類型：pyspark.rdd.PipelinedRDD通過hiveContext返回

RDD類型：pyspark.rdd.RDD

FPGrowth似乎只與PipelinedRDD工作。有什麼方法可以將常規RDD轉換爲PipelinedRDD？

謝謝！

Answer 1

好了，我的查詢是錯誤的，但改變了使用collect_set然後 我設法做繞過錯誤類型：

data=data.map(lambda row: row[0])