Pyspark Py4j IllegalArgumentException 与 spark.createDataFrame 和 pyspark.ml.clustering答案

【问题标题】：Pyspark Py4j IllegalArgumentException with spark.createDataFrame and pyspark.ml.clusteringPyspark Py4j IllegalArgumentException 与 spark.createDataFrame 和 pyspark.ml.clustering
【发布时间】：2023-04-09 14:55:01
【问题描述】：

让我先公开我的问题的完整背景，我将有一个简化的 MWE，它在底部重现相同的问题。随意跳过我对我的设置的漫谈，直接进入最后一部分。

我的原始问题中的参与者：

从 Amazon S3 读取的 spark 数据帧 data，其列 scaled_features 最终是 VectorAssembler 操作后跟 MinMaxScaler 的结果。
一个 spark 数据框列 pca_features，它是由上述 df 列在 PCA 之后产生的，如下所示：

mat = RowMatrix(data.select('scaled_features').rdd.map(list))
pc = mat.computePrincipalComponents(2)
projected = mat.multiply(pc).rows.map(lambda x: (x, )).toDF().withColumnRenamed('_1', 'pca_features')

BisectingKMeans 的两个实例适合上述数据帧中的两个特征实例，如下所示：

kmeans_scaled = BisectingKMeans(featuresCol='scaled_features').setK(4).setSeed(1)
model1 = kmeans_scaled.fit(data)

kmeans_pca = BisectingKMeans(featuresCol='pca_features').setK(4).setSeed(1)
model2 = kmeans_pca.fit(projected)

问题：

虽然 BisectingKMeans 从我的第一个 df 拟合到 scaled_features 没有问题，但在尝试拟合投影功能时，会出现以下错误

Py4JJavaError: An error occurred while calling o1413.fit.
: java.lang.IllegalArgumentException: requirement failed: Column features must be of type equal to one of the following types:
[struct<type:tinyint,size:int,indices:array<int>,values:array<double>>, array<double>, array<float>]
but was actually of type struct<type:tinyint,size:int,indices:array<int>,values:array<double>>.

如您所见，Py4J 抱怨我正在传递某种结构类型的数据，而该结构类型恰好是允许类型列表中指定的第一个类型。

其他调试信息：

我的 Spark 运行的是 2.4.0 版

检查 dtypes 产生：data.dtypes: [('scaled_features', 'vector')] 和 projected.dtypes: [('pca_features', 'vector')]。两种数据帧的 Schema 也是相同的，只打印一个以供参考：

root
 |-- scaled_features: vector (nullable = true)

重新创建错误 (MWE)：

事实证明，可以通过从一些向量创建一个简单的数据框来重新创建同样的错误（我原来的 dfs 中的列也是 VectorType）：

from pyspark.sql import Row
from pyspark.mllib.linalg import DenseVector
from pyspark.ml.clustering import BisectingKMeans

test_data = spark.createDataFrame([Row(test_features=DenseVector([43.0, 0.0, 200.0, 1.0, 1.0, 1.0, 0.0, 3.0])),
    Row(test_features=DenseVector([44.0, 0.0, 250.0, 1.0, 1.0, 1.0, 0.0, 1.0])),
    Row(test_features=DenseVector([23.0, 0.0, 0.0, 1.0, 1.0, 1.0, 0.0, 1.0])),
    Row(test_features=DenseVector([25.0, 0.0, 0.0, 1.0, 1.0, 1.0, 0.0, 2.0])),
    Row(test_features=DenseVector([19.0, 0.0, 200.0, 1.0, 0.0, 1.0, 0.0, 1.0]))])

kmeans_test = BisectingKMeans(featuresCol='test_features').setK(4).setSeed(1)
model3 = kmeans_test.fit(test_data)

最后一行导致我在原始设置中遇到的相同错误。

谁能解释这个错误并提出纠正它的方法？

【问题讨论】：

标签： python apache-spark pyspark apache-spark-mllib py4j

【解决方案1】：

经过几天的调查，我被指出了问题的（相当尴尬的）原因：

Pyspark 有两个机器学习库：pyspark.ml 和 pyspark.mllib，事实证明它们不能很好地结合在一起。将from pyspark.mllib.linalg import DenseVector 替换为from pyspark.ml.linalg import DenseVector 可以解决所有问题。

【讨论】：