pyspark SparseVectors 数据帧列 .dot 产品或使用 @udf 或 @pandas_udf 的任何其他向量类型列计算

Question

我确实尝试计算给定数据帧的两列之间的.dot乘积， SparseVectors 已经在 Spark 中具备了这种能力，所以我尝试以简单的方式执行此功能。可扩展的方式，无需转换为RDD或 DenseVectors 但我被困住了，花了过去 3 天的时间尝试找出一个 方法并失败，不返回传递的 2 个向量的计算 数据框中的列并寻找有关此事的指导， 拜托，因为我在这里缺少一些东西，不确定根本原因是什么......

对于单独的向量和 rdd 向量，此方法有效，但会失败 在传递数据帧列向量时工作，复制流程 和问题请参见下文，理想情况下此计算是并行发生的，因为实际工作数据具有数十亿或更多行（数据帧观察）：

from pyspark.ml.linalg import Vectors, SparseVector
    from pyspark.sql import Row
    df = spark.createDataFrame(
        [
         [["a","b","c"], SparseVector(4527, {0:0.6363067860791387, 1:1.0888040725098247, 31:4.371858972705023}),SparseVector(4527, {0:0.6363067860791387, 1:2.0888040725098247, 31:4.371858972705023})],
         [["d"], SparseVector(4527, {8: 2.729945780576634}), SparseVector(4527, {8: 4.729945780576634})],
        ], ["word", "i", "j"])

# # daframe content
df.show()
+---------+--------------------+--------------------+
|     word|                   i|                   j|
+---------+--------------------+--------------------+
|[a, b, c]|(4527,[0,1,31],[0...|(4527,[0,1,31],[0...|
|      [d]|(4527,[8],[2.7299...|(4527,[8],[4.7299...|
+---------+--------------------+--------------------+


@udf(returnType=ArrayType(FloatType()))
def sim_cos(v1, v2):
    if v1 is not None and v2 is not None:
        return float(v1.dot(v2))

# # calling udf
df = df.withColumn("dotP", sim_cos(df.i, df.j))

# # output after udf 
df.show()
+---------+--------------------+--------------------+----------+
|     word|                   i|                   j|      dotP|
+---------+--------------------+--------------------+----------+
|[a, b, c]|(4527,[0,1,31],[0...|(4527,[0,1,31],[0...|      null|
|      [d]|(4527,[8],[2.7299...|(4527,[8],[4.7299...|      null|
+---------+--------------------+--------------------+----------+

Answer 1

将 udf 重写为 lambda 可以在 Spark 2.4.5 上运行。发帖以防万一
任何人都对 PySpark 数据帧的这种方法感兴趣：

# # rewrite udf as lambda function:
sim_cos = F.udf(lambda x,y : float(x.dot(y)), FloatType())

# # executing udf on dataframe
df = df.withColumn("similarity", sim_cos(col("i"),col("j")))

# # end result
df.show()

+---------+--------------------+--------------------+----------+
|     word|                   i|                   j|similarity|
+---------+--------------------+--------------------+----------+
|[a, b, c]|(4527,[0,1,31],[0...|(4527,[0,1,31],[0...| 21.792336|
|      [d]|(4527,[8],[2.7299...|(4527,[8],[4.7299...| 12.912496|
+---------+--------------------+--------------------+----------+