使用 PySpark 进行主成分分析

Posted 2023-04-18

【中文标题】使用 PySpark 进行主成分分析

【英文标题】：Pca analysis with PySpark

【发布时间】：2017-02-24 10:02:44

【问题描述】：

我正在使用 PySpark 作为工具进行 PCA 分析，但由于从 csv 文件读取的数据的兼容性，我遇到了错误。我该怎么办？你能帮帮我吗？

from __future__ import print_function
from pyspark.ml.feature import PCA
from pyspark.ml.linalg import Vectors, VectorUDT

from pyspark.sql import SparkSession
from pyspark import SparkConf, SparkContext
from pyspark.sql.functions import udf
import pandas as pd
import numpy as np
from numpy import array


conf = SparkConf().setAppName("building a warehouse")
sc = SparkContext(conf=conf)

if __name__ == "__main__":
    spark = SparkSession\
        .builder\
        .appName("PCAExample")\
        .getOrCreate()



   data = sc.textFile('dataset.csv') \
        .map(lambda line:  line.split(','))\
        .collect()
   #create a data frame from data read from csv file 
   df = spark.createDataFrame(data, ["features"])
   #convert data to vector udt

   df.show()


   pca = PCA(k=3, inputCol="features", outputCol="pcaFeatures")
   model = pca.fit(df)

   result =  model.transform(df).select("pcaFeatures")
   result.show(truncate=False)

   spark.stop()

这是我得到的错误：

File "C:/spark/spark-2.1.0-bin-hadoop2.7/bin/pca_bigdata.py", line 38, in       <module>
model = pca.fit(df)
pyspark.sql.utils.IllegalArgumentException: u'requirement failed: Column features must be of type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 but was actually StringType.'

【问题讨论】：

你能提供一个文件的例子吗？谢谢。

它包含如下数据： 08636474609375,514,4246826171875,571,90142822265625,573,742431640625,586,60888671875,571,6429443359375,,

你的数字仍然被读取为字符串而不是浮点数，像这样做地图：data = sc.textFile('dataset.csv').map(lambda line: [float(k) for k in line.split(',')])

我试过你的指令，但我仍然在包含 model = pca.fit(df): u'requirement failed: Column features must be of type org.apache.spark.ml 的行上收到错误.linalg.VectorUDT@3bfc3ba7 但实际上是 DoubleType。'

@MehdiBenHamida 您需要将列类型 StringType 更改为 VectorUDT

【参考方案1】：

这里的错误指定自己的列需要VectorUDT 而不是StringType。所以这对你有用：-

from pyspark.mllib.linalg import SparseVector, VectorUDT       
from pyspark.sql.types import StringType, StructField, StructType
df = spark.createDataFrame(data, StructType([
                         StructField("features", VectorUDT(), True)
                       ]))