如何从 PySpark 中的向量列中提取浮点数？

Posted 2023-04-15

【中文标题】如何从 PySpark 中的向量列中提取浮点数？

【英文标题】：How to extract floats from vector columns in PySpark?

【发布时间】：2020-02-18 18:51:56

【问题描述】：

我的 Spark DataFrame 有以下格式的数据：

printSchema() 表明每一列的类型为vector。

我尝试使用下面的代码从 [ 和 ] 中获取值（对于 1 列 col1）：

from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType

firstelement=udf(lambda v:float(v[0]),FloatType())
df.select(firstelement('col1')).show()

但是，如何将其应用于df 的所有列？

【问题讨论】：

嗯...您正试图将 DenseVector 对象提供给模型，我从错误消息中想象。您应该提供纯 numpy 数组。您从哪里获取输入数据？

@DanielMöller：谢谢。正如我所提到的，x_train 和 x_test 是 pandas 数据帧。我找不到任何对 DenseVector 的引用...

它们应该是 numpy 数组，而不是数据帧。 x_train_numpy = x_train.values.

@DanielMöller：正如你所展示的，我将它们更改为 numpy。同样的错误。

那么数据框的格式可能不是很好。您可能应该研究一下并自己准备数据。

【参考方案1】：

1。提取单个向量列的第一个元素：

要获取向量列的第一个元素，您可以使用此 SO 中的答案：讨论Access element of a vector in a Spark DataFrame (Logistic Regression probability vector)

这是一个可重现的例子：

>>> from pyspark.sql import functions as f
>>> from pyspark.sql.types import FloatType
>>> df = spark.createDataFrame(["col1": [0.2], "col2": [0.25],
                                "col1": [0.45], "col2":[0.85]])
>>> df.show()
+------+------+
|  col1|  col2|
+------+------+
| [0.2]|[0.25]|
|[0.45]|[0.85]|
+------+------+

>>> firstelement=f.udf(lambda v:float(v[0]),FloatType())
>>> df.withColumn("col1", firstelement("col1")).show()
+----+------+
|col1|  col2|
+----+------+
| 0.2|[0.25]|
|0.45|[0.85]|
+----+------+

---

2。提取多个向量列的第一个元素：

要将上述解决方案推广到多列，请应用for loop。这是一个例子：

>>> from pyspark.sql import functions as f
>>> from pyspark.sql.types import FloatType

>>> df = spark.createDataFrame(["col1": [0.2], "col2": [0.25],
                                "col1": [0.45], "col2":[0.85]])
>>> df.show()
+------+------+
|  col1|  col2|
+------+------+
| [0.2]|[0.25]|
|[0.45]|[0.85]|
+------+------+

>>> firstelement=f.udf(lambda v:float(v[0]),FloatType())
>>> df = df.select([firstelement(c).alias(c) for c in df.columns])
>>> df.show()
+----+----+
|col1|col2|
+----+----+
| 0.2|0.25|
|0.45|0.85|
+----+----+

【讨论】：

太棒了！谢谢！

【参考方案2】：

据我了解，您不需要使用UDF 将 Vector 更改为正常的 Float 类型。使用pyspark预定义函数concat_ws。

>>> from pyspark.sql.functions import *
>>> df.show()
+------+
|   num|
+------+
| [211]|
|[3412]|
| [121]|
| [121]|
|  [34]|
|[1441]|
+------+

>>> df.printSchema()
root
 |-- num: array (nullable = true)
 |    |-- element: string (containsNull = true)

>>> df.withColumn("num", concat_ws("", col("num"))).show()
+----+
| num|
+----+
| 211|
|3412|
| 121|
| 121|
|  34|
|1441|
+----+