PySpark 上的 PCA 不规则执行答案

【问题标题】：PCA on PySpark irregular executionPySpark 上的 PCA 不规则执行
【发布时间】：2017-03-15 12:47:44
【问题描述】：

我正在使用 PySpark 通过 csv 文件处理 PCA。我有一些奇怪的行为；我的代码有时可以完美运行，但有时会返回此错误：

 File "C:/spark/spark-2.1.0-bin-hadoop2.7/bin/pca_final2.py", line 25, in <module>
columns = (fileObj.first()).split(';')
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\rdd.py", line 1361, in first
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\rdd.py", line 1343, in take
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\context.py", line 965, in runJob
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\py4j-0.10.4-src.zip\py4j\java_gateway.py", line 1133, in __call__
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\utils.py", line 63, in deco
File "C:\spark\spark-2.1.0-bin-hadoop2.7\python\lib\py4j-0.10.4-src.zip\py4j\protocol.py", line 319, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, localhost, executor driver): java.net.SocketException: Connection reset by peer: socket write error

这是我的代码：

#########################! importing libraries !########################
from __future__ import print_function
from pyspark.ml.linalg import Vectors
from pyspark.sql import SparkSession
from pyspark import SparkConf, SparkContext
from pyspark.ml.feature import PCA, VectorAssembler
from pyspark.mllib.linalg import Vectors
from pyspark.ml import Pipeline
from pyspark.sql import SQLContext
from pyspark import SparkContext
from pyspark.mllib.feature import Normalizer
import timeit
########################! main script !#################################
sc = SparkContext("local", "pca-app")
sqlContext = SQLContext(sc)
if __name__ == "__main__":
    spark = SparkSession\
        .builder\
        .appName("PCAExample")\
        .getOrCreate()  
    start=timeit.default_timer() 
    fileObj = sc.textFile('bigiris.csv')
    data = fileObj.map(lambda line: [float(k) for k in line.split(';')])
    columns = (fileObj.first()).split(';')
    df = spark.createDataFrame(data, columns)
    df.show()
    vecAssembler = VectorAssembler(inputCols=columns, outputCol="features")
    pca = PCA(k=2, inputCol="features", outputCol="pcaFeatures")
    pipeline = Pipeline(stages=[vecAssembler, pca])
    model = pipeline.fit(df)
    result = model.transform(df).select("pcaFeatures")
    stop=timeit.default_timer()
    result.show(truncate=False)
    time=stop-start
    print ("this operation takes ", (time), " seconds")
    spark.stop()

为什么我会得到这种不正常的执行？以及我应该添加什么来解决这个问题。

【问题讨论】：

你能给我们bigiris.csv的前几行吗？
1;2;3;4;5;6;7;8;9;10;11;12;13;14;15;16;17;18;19;20;21; 22;23;24;25;26;27;28;29;30;31;32;33;34;35;36;37;38;39;40;41;42;43;44;45;46; 47;48;49;50
标题看起来一样吗？
是的，所有文件都具有相同的结构

标签： csv pyspark pca

【解决方案1】：

创建data 框架时，您没有过滤掉标题。假设您的列名是字符串，这将导致错误，因为列名无法转换为浮点值。请参阅下面的脚本修改部分，该部分使用 filter 删除标题。

fileObj = sc.textFile('e:/iris.data.txt')
header = fileObj.first()
data = fileObj.filter(lambda x: x != header).map(lambda line: [float(k) for k in line.split(';')])
columns = header.split(';')
df = spark.createDataFrame(data, columns)
df.show()

【讨论】：

【解决方案2】：

此处错误通知在columns = (fileObj.first()).split(';') 行。基本上你是在(;)的基础上尝试splitfileObj的第一行。这里执行的操作顺序是错误的，因为 line 已经在上一步中转换为 list。

正确的操作顺序是这样的（columns line should be before data line）：-

fileObj = sc.textFile('bigiris.csv')
columns = (fileObj.first()).split(';')
data = fileObj.map(lambda line: [float(k) for k in line.split(';')])
df = spark.createDataFrame(data, columns)

错误原因：- 行 (data =) 有 fileObj.map 和 line.split(';') 。已经将 csv 的每一行相对于(;) 分开了

如果您在 csv 中将标题作为文本并希望从数据中删除，请按照 Jaco 回答 filter(lambda x: x != header)。

【讨论】：