PySpark 中的广播随机森林模型答案

【问题标题】：Broadcast Random-Forest Model in PySparkPySpark 中的广播随机森林模型
【发布时间】：2015-11-11 12:54:39
【问题描述】：

我使用的是 spark 1.4.1。当我尝试广播随机森林模型时，它向我显示此错误：

Traceback (most recent call last):
  File "/gpfs/haifa/home/d/a/davidbi/codeBook/Nice.py", line 358, in <module>
broadModel = sc.broadcast(model)
  File "/opt/apache/spark-1.4.1-bin-hadoop2.4_doop/python/lib/pyspark.zip/pyspark/context.py", line 698, in broadcast
  File "/opt/apache/spark-1.4.1-bin-hadoop2.4_doop/python/lib/pyspark.zip/pyspark/broadcast.py", line 70, in __init__
  File "/opt/apache/spark-1.4.1-bin-hadoop2.4_doop/python/lib/pyspark.zip/pyspark/broadcast.py", line 78, in dump
File "/opt/apache/spark-1.4.1-bin-hadoop2.4_doop/python/lib/pyspark.zip/pyspark/context.py", line 252, in __getnewargs__
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.

我尝试执行的代码示例：

sc = SparkContext(appName= "Something")
model = RandomForest.trainRegressor(sc.parallelize(data), categoricalFeaturesInfo=categorical, numTrees=100, featureSubsetStrategy="auto", impurity='variance', maxDepth=4)
broadModel= sc.broadcast(model)

如果有人可以帮助我，我将非常感激！非常感谢！

【问题讨论】：

是否有需要广播整个模型的原因？该模型可以对输入的 RDD 进行预测。
有不止一个模型（在我的例子中，每个模型定义组）。每个样本都需要从每个模型中得到预测，才能知道他最喜欢哪个组。我正在处理大数据，所以我需要将模型广播到映射器。

标签： apache-spark pyspark broadcast random-forest apache-spark-mllib

【解决方案1】：

简短的回答是使用 PySpark 是不可能的。预测所需的callJavaFunc 使用的是SparkContext，因此会出现错误。不过，使用 Scala API 可以做这样的事情。

在 Python 中，您可以使用与单个模型相同的方法，这意味着 model.predict 后跟 zip。

models = [mode1, mode2, mode3]

predictions = [
    model.predict(testData.map(lambda x: x.features)) for model in models]

def flatten(x):
    if isinstance(x[0], tuple):
        return tuple(list(x[0]) + [x[1]])
    else:
        return x

(testData
   .map(lambda lp: lp.label)
   .zip(reduce(lambda p1, p2: p1.zip(p2).map(flatten), predictions)))

如果想了解更多问题的根源，请查看：How to use Java/Scala function from an action or a transformation?

【讨论】：