“PipelinedRDD”对象在PySp中没有属性“toDF”

Question

我正在尝试加载一个SVM文件并将其转换为一个DataFrame，这样我就可以使用Spark中的ML模块（PipelineML）。 我刚刚在Ubuntu 14.04上安装了一个新的Spark 1.5.0（没有配置spark-env.sh）。

我的my_script.py是：

from pyspark.mllib.util import MLUtils
from pyspark import SparkContext

sc = SparkContext("local", "Teste Original")
data = MLUtils.loadLibSVMFile(sc, "/home/svm_capture").toDF()

我用的是：./spark-submit my_script.py

我得到了错误：

Traceback (most recent call last):
File "/home/fred-spark/spark-1.5.0-bin-hadoop2.6/pipeline_teste_original.py", line 34, in <module>
data = MLUtils.loadLibSVMFile(sc, "/home/fred-spark/svm_capture").toDF()
AttributeError: 'PipelinedRDD' object has no attribute 'toDF'

我不能理解的是，如果我跑：

data = MLUtils.loadLibSVMFile(sc, "/home/svm_capture").toDF()

直接在PySpark外壳内部，它工作。

Answer 1

toDF方法是一个猴子补丁executed inside ^{} (^{} constructor in 1.x) constructor，因此要使用它，必须首先创建一个SQLContext（或SparkSession）：

# SQLContext or HiveContext in Spark 1.x
from pyspark.sql import SparkSession
from pyspark import SparkContext

sc = SparkContext()

rdd = sc.parallelize([("a", 1)])
hasattr(rdd, "toDF")
## False

spark = SparkSession(sc)
hasattr(rdd, "toDF")
## True

rdd.toDF().show()
## +---+---+
## | _1| _2|
## +---+---+
## |  a|  1|
## +---+---+

更不用说你首先需要一个SQLContext或SparkSession来处理DataFrames。