问题的简短版本！
请考虑以下代码片段(假设spark已设置为某些SparkSession)：
from pyspark.sql import Row
source_data = [
Row(city="Chicago", temperatures=[-1.0, -2.0, -3.0]),
Row(city="New York", temperatures=[-7.0, -7.0, -5.0]),
]
df = spark.createDataFrame(source_data)
请注意，温度字段是浮点数的列表。我想将这些浮点数列表转换为 MLlib 类型Vector，我希望使用基本的DataFrameAPI 而不是通过 RDD 来表达这种转换(这是低效的，因为它将所有数据从 JVM 发送到 Python，处理是在 Python 中完成的，我们没有得到 Spark 的 Catalyst 优化器的好处， 亚达亚达)。我该怎么做？具体说来：
- 有没有办法让直接演员工作？有关详细信息(以及解决方法的失败尝试)，请参阅下文？或者，是否有任何其他操作具有我所追求的效果？
- 在我下面建议的两个替代解决方案中，哪个更有效(UDF 与分解/重新组装列表中的项目)？还是有没有其他几乎但不完全正确的替代方案比它们中的任何一个更好？
直投不起作用
这就是我所期望的"适当"解决方案。我想将列的类型从一种类型转换为另一种类型，所以我应该使用强制转换。作为一点上下文，让我提醒您将其转换为另一种类型的正常方法：
from pyspark.sql import types
df_with_strings = df.select(
df["city"],
df["temperatures"].cast(types.ArrayType(types.StringType()))),
)
现在例如df_with_strings.collect()[0]["temperatures"][1]'-7.0'.但是如果我投射到 ml 矢量，那么事情就不那么顺利了：
from pyspark.ml.linalg import VectorUDT
df_with_vectors = df.select(df["city"], df["temperatures"].cast(VectorUDT()))
这会产生一个错误：
pyspark.sql.utils.AnalysisException: "cannot resolve 'CAST(`temperatures` AS STRUCT<`type`: TINYINT, `size`: INT, `indices`: ARRAY<INT>, `values`: ARRAY<DOUBLE>>)' due to data type mismatch: cannot cast ArrayType(DoubleType,true) to org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7;;
'Project [city#0, unresolvedalias(cast(temperatures#1 as vector), None)]
+- LogicalRDD [city#0, temperatures#1]
"
哎呀！有什么想法可以解决这个问题吗？
可能的替代方案
备选方案1：使用VectorAssembler
有一种Transformer似乎几乎非常适合这项工作：VectorAssembler。它采用一列或多列，并将它们连接成单个向量。不幸的是，它只需要Vector和Float列，而不是Array列，因此以下内容不起作用：
from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=["temperatures"], outputCol="temperature_vector")
df_fail = assembler.transform(df)
它给出了此错误：
pyspark.sql.utils.IllegalArgumentException: 'Data type ArrayType(DoubleType,true) is not supported.'
我能想到的最好的解决方法是将列表分解为多列，然后使用VectorAssembler再次将它们全部收集起来：
from pyspark.ml.feature import VectorAssembler
TEMPERATURE_COUNT = 3
assembler_exploded = VectorAssembler(
inputCols=["temperatures[{}]".format(i) for i in range(TEMPERATURE_COUNT)],
outputCol="temperature_vector"
)
df_exploded = df.select(
df["city"],
*[df["temperatures"][i] for i in range(TEMPERATURE_COUNT)]
)
converted_df = assembler_exploded.transform(df_exploded)
final_df = converted_df.select("city", "temperature_vector")
这似乎是理想的，除了TEMPERATURE_COUNT超过 100，有时超过 1000。(另一个问题是，如果您事先不知道数组的大小，代码会更复杂，尽管我的数据并非如此。Spark 是否实际上生成了一个包含这么多列的中间数据集，或者它只是认为这是单个项目暂时通过的中间步骤(或者当它看到这些列的唯一用途是组装成一个向量时，它是否完全优化了这个离开步骤)？
备选方案 2：使用 UDF
一个更简单的替代方法是使用 UDF 进行转换。这让我可以在一行代码中非常直接地表达我想做什么，并且不需要创建一个具有大量列的数据集。但是所有这些数据都必须在Python和JVM之间交换，并且每个单独的数字都必须由Python处理(众所周知，这对于迭代单个数据项来说非常慢)。这是它的外观：
from pyspark.ml.linalg import Vectors, VectorUDT
from pyspark.sql.functions import udf
list_to_vector_udf = udf(lambda l: Vectors.dense(l), VectorUDT())
df_with_vectors = df.select(
df["city"],
list_to_vector_udf(df["temperatures"]).alias("temperatures")
)
可忽略的备注
这个漫无边际的问题的其余部分是我在试图找到答案时想出的一些额外内容。大多数阅读本文的人可能会跳过它们。
不是解决方案：首先使用Vector
在这个简单的示例中，可以使用向量类型创建数据，但当然，我的数据并不是我正在并行化的 Python 列表，而是从数据源读取的。但为了记录，这是看起来的样子：
from pyspark.ml.linalg import Vectors
from pyspark.sql import Row
source_data = [
Row(city="Chicago", temperatures=Vectors.dense([-1.0, -2.0, -3.0])),
Row(city="New York", temperatures=Vectors.dense([-7.0, -7.0, -5.0])),
]
df = spark.createDataFrame(source_data)
低效解决方案：使用map()
一种可能性是使用 RDDmap()方法将列表转换为Vector。这与 UDF 的想法类似，只是它更糟，因为每行中的所有字段都会产生序列化等成本，而不仅仅是正在操作的字段。作为记录，该解决方案如下所示：
df_with_vectors = df.rdd.map(lambda row: Row(
city=row["city"],
temperatures=Vectors.dense(row["temperatures"])
)).toDF()
尝试转换解决方法失败
无奈之下，我注意到Vector在内部由具有四个字段的结构表示，但是使用该类型结构的传统强制转换也不起作用。这是一个插图(我使用 udf 构建了结构，但 udf 不是重要的部分)：
from pyspark.ml.linalg import Vectors, VectorUDT
from pyspark.sql.functions import udf
list_to_almost_vector_udf = udf(lambda l: (1, None, None, l), VectorUDT.sqlType())
df_almost_vector = df.select(
df["city"],
list_to_almost_vector_udf(df["temperatures"]).alias("temperatures")
)
df_with_vectors = df_almost_vector.select(
df_almost_vector["city"],
df_almost_vector["temperatures"].cast(VectorUDT())
)
这给出了错误：
pyspark.sql.utils.AnalysisException: "cannot resolve 'CAST(`temperatures` AS STRUCT<`type`: TINYINT, `size`: INT, `indices`: ARRAY<INT>, `values`: ARRAY<DOUBLE>>)' due to data type mismatch: cannot cast StructType(StructField(type,ByteType,false), StructField(size,IntegerType,true), StructField(indices,ArrayType(IntegerType,false),true), StructField(values,ArrayType(DoubleType,false),true)) to org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7;;
'Project [city#0, unresolvedalias(cast(temperatures#5 as vector), None)]
+- Project [city#0, <lambda>(temperatures#1) AS temperatures#5]
+- LogicalRDD [city#0, temperatures#1]
"
就我个人而言，我会使用 Python UDF，不会打扰其他任何事情：
Vectors不是本机 SQL 类型，因此会以某种方式产生性能开销。特别是，此过程需要两个步骤，其中首先将数据从外部类型转换为行，然后使用泛型RowEncoder从行转换为内部表示。- 任何下游 ML
Pipeline都将比简单的转换昂贵得多。此外，它需要一个与上述相反的过程
但是，如果您真的想要其他选择，则可以：
-
带有 Python 包装器的 Scala UDF：
按照项目网站上的说明安装 sbt。
创建具有以下结构的 Scala 包：
. ├── build.sbt └── udfs.scala编辑
build.sbt(调整以反映 Scala 和 Spark 版本)：scalaVersion := "2.11.8" libraryDependencies ++= Seq( "org.apache.spark" %% "spark-sql" % "2.4.4", "org.apache.spark" %% "spark-mllib" % "2.4.4" )编辑
udfs.scala：package com.example.spark.udfs import org.apache.spark.sql.functions.udf import org.apache.spark.ml.linalg.DenseVector object udfs { val as_vector = udf((xs: Seq[Double]) => new DenseVector(xs.toArray)) }包：
sbt package并包括(或等效，取决于 Scala 版本)：
$PROJECT_ROOT/target/scala-2.11/udfs_2.11-0.1-SNAPSHOT.jar作为启动壳牌/提交申请时
--driver-class-path的论据。在 PySpark 中定义一个包装器：
from pyspark.sql.column import _to_java_column, _to_seq, Column from pyspark import SparkContext def as_vector(col): sc = SparkContext.getOrCreate() f = sc._jvm.com.example.spark.udfs.udfs.as_vector() return Column(f.apply(_to_seq(sc, [col], _to_java_column)))测试：
with_vec = df.withColumn("vector", as_vector("temperatures")) with_vec.show()+--------+------------------+----------------+ | city| temperatures| vector| +--------+------------------+----------------+ | Chicago|[-1.0, -2.0, -3.0]|[-1.0,-2.0,-3.0]| |New York|[-7.0, -7.0, -5.0]|[-7.0,-7.0,-5.0]| +--------+------------------+----------------+ with_vec.printSchema()root |-- city: string (nullable = true) |-- temperatures: array (nullable = true) | |-- element: double (containsNull = true) |-- vector: vector (nullable = true)
将数据 转储为反映
DenseVector架构的 JSON 格式并读回：from pyspark.sql.functions import to_json, from_json, col, struct, lit from pyspark.sql.types import StructType, StructField from pyspark.ml.linalg import VectorUDT json_vec = to_json(struct(struct( lit(1).alias("type"), # type 1 is dense, type 0 is sparse col("temperatures").alias("values") ).alias("v"))) schema = StructType([StructField("v", VectorUDT())]) with_parsed_vector = df.withColumn( "parsed_vector", from_json(json_vec, schema).getItem("v") ) with_parsed_vector.show()+--------+------------------+----------------+ | city| temperatures| parsed_vector| +--------+------------------+----------------+ | Chicago|[-1.0, -2.0, -3.0]|[-1.0,-2.0,-3.0]| |New York|[-7.0, -7.0, -5.0]|[-7.0,-7.0,-5.0]| +--------+------------------+----------------+with_parsed_vector.printSchema()root |-- city: string (nullable = true) |-- temperatures: array (nullable = true) | |-- element: double (containsNull = true) |-- parsed_vector: vector (nullable = true)
我和你一样有同样的问题，我这样做了。 这种方式包括RDD转换，因此对性能不至关重要，但它是有效的。
from pyspark.sql import Row
from pyspark.ml.linalg import Vectors
source_data = [
Row(city="Chicago", temperatures=[-1.0, -2.0, -3.0]),
Row(city="New York", temperatures=[-7.0, -7.0, -5.0]),
]
df = spark.createDataFrame(source_data)
city_rdd = df.rdd.map(lambda row:row[0])
temp_rdd = df.rdd.map(lambda row:row[1])
new_df = city_rdd.zip(temp_rdd.map(lambda x:Vectors.dense(x))).toDF(schema=['city','temperatures'])
new_df
结果是，
DataFrame[city: string, temperatures: vector]
For pyspark>=3.1.0
从 3.1.0 开始，有一个构建解决方案：array_to_vector。
鉴于您的情况：
from pyspark.ml.functions import vector_to_array
df = df.withColumn("temperatures_vectorized", vector_to_array("temperatures"))
PS：同样从 3.0.0 开始，还有一个相反的操作：vector_to_array