'PipelinedRDD' 对象在 PySpark 中没有属性 'toDF'

Question

我正在尝试加载 SVM 文件并将其转换为 DataFrame，以便我可以使用 Spark 的 ML 模块 (Pipeline ML)。我刚刚在 Ubuntu 14.04（未配置 spark-env.sh）上安装了新的 Spark 1.5.0。

我的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 中 shell，它有效。

Answer 1

toDF 方法是一个猴子补丁 executed inside SparkSession (SQLContext 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。

Answer 2

确保你也有 spark 会话。

sc = SparkContext("local", "first app")
spark = SparkSession(sc)

'PipelinedRDD' 对象在 PySpark 中没有属性 'toDF'

'PipelinedRDD' object has no attribute 'toDF' in PySpark

python

apache-spark

rdd

apache-spark-sql

pyspark