尝试根据两个 ArrayType 列的差异创建新列

Question

我有一张这样的表：

+-----+----+-------+-------+
|name | id | msg_a | msg_b |
+-----+----+-------+-------+
|    a|   3|[a,b,c]|[c]    |
|    b|   5|[x,y,z]|[h,x,z]|
|    c|   7|[a,x,y]|[j,x,y]|
+-----+----+-------+-------+

我想添加一个列，以便显示msg_b而不是msg_a中的任何内容。例如

+-----+----+-------+-------+------------+
|name | id | msg_a | msg_b | difference |
+-----+----+-------+-------+------------+
|    a|   3|[a,b,c]|[c]    |NA          |
|    b|   5|[x,y,z]|[h,x,z]|[h]         |
|    c|   7|[a,x,y]|[j,x,y]|[j]         |
+-----+----+-------+-------+------------+

参考以前的帖子，我已经尝试过 df.select('msg_b').subtract(df.select('msg_a')).show()这有效，但我需要信息作为表格，带有name和id

这样做： df.withColumn("difference", F.col('msg_b').subtract(F.col(''msg_a'))).show(5)产生TypeError: 'Column' object is not callable

不确定是否有单独的功能来执行此操作，如果我错过了一些明显的东西等。

Answer 1

你必须使用 UDF ：

from pyspark.sql.functions import *
from pyspark.sql.types import *
@udf(ArrayType(StringType()))
def subtract(xs, ys):
    return list(set(xs) - set(ys))

例

df = sc.parallelize([
   (["a", "b", "c"], ["c"]), (["x", "y", "z"], ["h", "x", "z"])
]).toDF(["msg_a", "msg_b"])
df.select(subtract('msg_b', 'msg_a'))
+----------------------+
|subtract(msg_b, msg_a)|
+----------------------+
|                    []|
|                   [h]|
+----------------------+