pyspark MLlib踩坑之model predict+rdd map zip,zip使用尤其注意啊啊啊!-阿里云开发者社区

开发者社区> 大数据> 正文

pyspark MLlib踩坑之model predict+rdd map zip,zip使用尤其注意啊啊啊!

简介:

一开始是因为没法直接在pyspark里使用map 来做model predict,但是scala是可以的!如下:

When we use Scala API a recommended way of getting predictions for RDD[LabeledPoint] using DecisionTreeModel is to simply map over RDD:

val labelAndPreds = testData.map { point =>
  val prediction = model.predict(point.features)
  (point.label, prediction)
}

Unfortunately similar approach in PySpark doesn't work so well:

labelsAndPredictions = testData.map(
    lambda lp: (lp.label, model.predict(lp.features))
labelsAndPredictions.first()

Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transforamtion. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.

Instead of that official documentation recommends something like this:

predictions = model.predict(testData.map(lambda x: x.features))
labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions)

 

而这就是万恶的根源,因为zip在某些情况下并不能得到你想要的结果,就是说zip后的顺序是混乱的!!!我就在项目里遇到了!!!

 

This appears to imply that even the trivial a.map(f).zip(a) is not guaranteed to be equivalent to a.map(x => (f(x),x)). What is the situation when zip() results are reproducible?

见:https://stackoverflow.com/questions/29268210/mind-blown-rdd-zip-method

原因:

zip is generally speaking a tricky operation. It requires both RDDs not only to have the same number of partitions but also the same number of elements per partition.

Excluding some special cases this is guaranteed only if both RDDs have the same ancestor and there are not shuffles and operations potentially changing number of elements (filterflatMap) between the common ancestor and the current state. Typically it means only map (1-to-1) transformations.

见:https://stackoverflow.com/questions/32084368/can-only-zip-with-rdd-which-has-the-same-number-of-partitions-error

 

根源是因为我的ancestor rdd做了shuffle和filter的操作!最后在他们的子rdd上使用zip就会出错(数据乱序了)!!!真是太郁闷了,折腾一天这个问题,感谢上帝终于解决了!阿门!

 

最后我的解决方法是:

1、直接将rdd做union操作,rdd = rdd.union(sc.parallelize([])),然后map,zip就能输出正常结果了!

2、或者是直接将预测的rdd collect到driver机器,使用model predict,是比较丑陋的做法!


 









本文转自张昺华-sky博客园博客,原文链接:http://www.cnblogs.com/bonelee/p/7218268.html,如需转载请自行联系原作者

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
大数据
使用钉钉扫一扫加入圈子
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

其他文章