寻找专业知识来指导我解决以下问题。
背景:
- 我正在尝试使用受启发的基本 PySpark 脚本这
例子 https://github.com/GoogleCloudPlatform/spark-recommendation-engine/blob/master/pyspark/app_collaborative.py
- 作为部署基础设施,我使用 Google Cloud Dataproc 集群。
- 我的代码中的基石是记录的函数“recommendProductsForUsers”here https://spark.apache.org/docs/1.4.0/api/java/org/apache/spark/mllib/recommendation/MatrixFactorizationModel.html#recommendProductsForUsers(int)这为我返回了模型中所有用户的前 X 个产品
我遇到的问题
ALS.Train 脚本在 GCP 上运行流畅且扩展良好(轻松超过 100 万客户)。
然而,应用预测:即使用函数“PredictAll”或“recommendProductsForUsers”,根本无法扩展。我的脚本对于小型数据集( 50k 客户和 > 10k 产品)
-
我得到的错误如下:
16/08/16 14:38:56 WARN org.apache.spark.scheduler.TaskSetManager:
Lost task 22.0 in stage 411.0 (TID 15139,
productrecommendation-high-w-2.c.main-nova-558.internal):
java.lang.StackOverflowError
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:373)
at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
at sun.reflect.GeneratedMethodAccessor11.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1058)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1909)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:373)
at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
我什至获得了一个 300 GB 集群(1 个 108 GB 主节点 + 2 个 108 GB RAM 节点)来尝试运行它;它适用于 50,000 名客户,但不适用于更多客户
我的目标是拥有一个可以为超过 80 万客户运行的设置
Details
失败的代码行
predictions = model.recommendProductsForUsers(10).flatMap(lambda p: p[1]).map(lambda p: (str(p[0]), str(p[1]), float(p[2])))
pprint.pprint(predictions.take(10))
schema = StructType([StructField("customer", StringType(), True), StructField("sku", StringType(), True), StructField("prediction", FloatType(), True)])
dfToSave = sqlContext.createDataFrame(predictions, schema).dropDuplicates()
您建议如何进行?我觉得脚本末尾的“合并”部分(即当我将其写入 dfToSave 时)导致了错误;有没有办法绕过这个并逐部分保存?
从堆栈跟踪来看,这似乎与以下问题相同使用 ALS 训练时 Spark 给出 StackOverflowError https://stackoverflow.com/questions/31484460/spark-gives-a-stackoverflowerror-when-training-using-als
基本上,Spark 以递归方式表达 RDD 谱系,因此,当在迭代工作负载过程中未对事物进行惰性评估时,您最终会得到深度嵌套的对象。调用 sc.setCheckpointDir 并调整检查点间隔将减少此 RDD 谱系的长度。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)