将 libsvm 格式字符串 ("field1:value field2:value") 转换为值的 DenseVector

Question

我有一个 libsvm 格式的专栏（火花的毫升库）field1:value field2:value ...

+--------------+-----+
|      features|label|
+--------------+-----+
|   a:1 b:2 c:3|    0|
|   a:4 b:5 c:6|    0|
|   a:7 b:8 c:9|    1|
|a:10 b:11 c:12|    0|
+--------------+-----+

我想提取值并将它们保存在 pyspark 中每一行的数组中

features.printSchema()

root
 |-- features: string (nullable = false)
 |-- label: integer (nullable = true)

我正在使用以下 udf，因为受影响的列是数据框的一部分

from pyspark.sql.functions import udf
from pyspark.ml.linalg import Vectors

features_expl = udf(lambda features: Vectors.dense(features.split(" ")).map(lambda feat: float(str(feat.split(":")[1]))))
features=features.withColumn("feats", features_expl(features.features))

我得到的结果是： ValueError：无法将字符串转换为浮点数：移动：0.0 它似乎没有执行第二次拆分并在字符串上调用 float()。

我想得到的是：

+--------------+-----+
|      features|label|
+--------------+-----+
|     [1, 2, 3]|    0|
|     [4, 5, 6]|    0|
|     [7, 8, 9]|    1|
|  [10, 11, 12]|    0|
+--------------+-----+

Answer 1

您的udf 存在两个主要问题。首先，它没有按您的预期工作。将代码的核心视为以下函数：

from pyspark.ml.linalg import Vectors
def features_expl_non_udf(features): 
    return Vectors.dense(
        features.split(" ")).map(lambda feat: float(str(feat.split(":")[1]))
    )

如果您在其中一个字符串上调用它：

features_expl_non_udf("a:1 b:2 c:3")
#ValueError: could not convert string to float: a:1

因为features.split(" ") 返回['a:1', 'b:2', 'c:3']，您将其传递给Vectors.dense 构造函数。这没有任何意义。

您打算首先在空间上拆分，然后在: 上拆分结果列表的每个值。然后您可以将这些值转换为float 并将列表传递给Vectors.dense。

这是您的逻辑的正确实现：

def features_expl_non_udf(features): 
    return Vectors.dense(map(lambda feat: float(feat.split(":")[1]), features.split()))
features_expl_non_udf("a:1 b:2 c:3")
#DenseVector([1.0, 2.0, 3.0])

现在您的udf 的第二个问题是您没有指定returnType。对于DenseVector，您需要use VectorUDT as the returnType。

from pyspark.sql.functions import udf
from pyspark.ml.linalg import VectorUDT

features_expl = udf(
    lambda features: Vectors.dense(
        map(lambda feat: float(feat.split(":")[1]), features.split())
    ),
    VectorUDT()
)
features.withColumn("feats", features_expl(features.features)).show()
#+--------------+-----+----------------+
#|      features|label|           feats|
#+--------------+-----+----------------+
#|   a:1 b:2 c:3|    0|   [1.0,2.0,3.0]|
#|   a:4 b:5 c:6|    0|   [4.0,5.0,6.0]|
#|   a:7 b:8 c:9|    1|   [7.0,8.0,9.0]|
#|a:10 b:11 c:12|    0|[10.0,11.0,12.0]|
#+--------------+-----+----------------+

---

作为替代方案，您可以使用regexp_replace 和split 在火花端执行字符串处理部分，但您仍然必须使用udf 将最终输出转换为DenseVector。

from pyspark.sql.functions import regexp_replace, split, udf
from pyspark.ml.linalg import Vectors, VectorUDT

toDenseVector = udf(Vectors.dense, VectorUDT())

features.withColumn(
    "features",
    toDenseVector(
        split(regexp_replace("features", r"\w+:", ""), "\s+").cast("array<float>")
    )
).show()
#+----------------+-----+
#|        features|label|
#+----------------+-----+
#|   [1.0,2.0,3.0]|    0|
#|   [4.0,5.0,6.0]|    0|
#|   [7.0,8.0,9.0]|    1|
#|[10.0,11.0,12.0]|    0|
#+----------------+-----+