我有一个非常简单的用例,但可能会有非常大的结果集。我的代码在pyspark shell上执行以下操作:
from pyspark.mllib.fpm import FPGrowth
data = sc.textFile("/Users/me/associationtestproject/data/sourcedata.txt")
transactions = data.map(lambda line: line.strip().split(' '))
model = FPGrowth.train(transactions, minSupport=0.000001, numPartitions=1000)
# Perform any RDD operation
for item in model.freqItemsets().toLocalIterator():
# do something with item
每当我通过调用count()或toLocalIterator来启动实际处理时,我的操作最终会出现内存不足错误。 FPGrowth没有将我的数据分区吗? 我的结果数据很大,即使是一个单个分区也会卡住我的内存吗? 如果是这样,有没有办法以“流”方式将RDD持久化到磁盘上,而不是试图将其保存在内存中?
感谢您提供任何见解。
编辑:FPGrowth的一个基本限制是整个FP树必须放入内存中。 因此,提高最小支持阈值的建议是有效的。
-Raj