我在将 SQL 数据框中的列元素相乘时遇到困难。
sv1 = Vectors.sparse(3, [0, 2], [1.0, 3.0])
sv2 = Vectors.sparse(3, [0, 1], [2.0, 4.0])
def xByY(x,y):
return np.multiply(x,y)
print(xByY(sv1, sv2))
以上作品。
但下面没有。
xByY_udf = udf(xByY)
tempDF = sqlContext.createDataFrame([(sv1, sv2), (sv1, sv2)], ('v1', 'v2'))
tempDF.show()
print(tempDF.select(xByY_udf('v1', 'v2')).show())
非常感谢!
请您参考如下方法:
如果您希望您的 udf 返回一个 SparseVector，我们首先需要修改您的函数的输出，然后设置 的输出模式>udf 到 VectorUDT():
要声明一个SparseVector，我们需要原始数组的大小，以及索引和值 非零 元素。如果乘法的中间结果是 list，我们可以使用 len() 和列表推导式找到它们:
from pyspark.ml.linalg import Vectors, VectorUDT
def xByY(x,y):
res = np.multiply(x,y).tolist()
vec_args = len(res), [i for i,x in enumerate(res) if x != 0], [x for x in res if x != 0]
return Vectors.sparse(*vec_args)
现在我们可以声明我们的 udf 并测试它了:
xByY_udf = udf(xByY, VectorUDT())
tempDF.select(xByY_udf('v1', 'v2')).show()
+-------------+
| xByY(v1, v2)|
+-------------+
|(3,[0],[2.0])|
|(3,[0],[2.0])|
+-------------+
