如何压缩两列，分解它们，最后旋转？

Question

我有两个数组列（names，score）。我要把它们都炸掉。将名称作为得分的name列（类似于pivot）

+------------+-------------------------+--------------------+                        
|      id    |     names               |      score         |
+------------+-------------------------+--------------------+
|ab01        |[F1 , F2, F3, F4, F5]    |[00123, 000.001, 00127, 00.0123, 111]
|ab02        |[F1 , F2, F3, F4, F5, F6]|[00124, 000.003, 00156, 00.067,  156, 254]
|ab03        |[F1 , F2, F3, F4, F5]    |[00234, 000.078, 00188, 00.0144, 188]
|ab04        |[F1 , F2, F3, F4, F5]    |[00345, 000.01112, 001567, 00.0186, 555]

预期产出：

 id       F1      F2        F3        F4    F5  F6
ab01    00123   000.001    00127    00.0123 111 null
ab02    00124   000.003    00156    00.067  156 254
ab03    00234   000.078    00188    00.0144 188 null
ab04    00345   000.01112  001567   00.0186 555 null

我试着把名字和分数拉上拉链，然后把它们炸开

combine = F.udf(lambda x, y: list(zip(x, y)),
                ArrayType(
                          StructType(
                                     [StructField("names", StringType()),
                                      StructField("score", StringType())
                                     ]
                                    )
                         )
               )

df2 = df.withColumn("new", combine("score", "names"))
         .withColumn("new", F.explode("new"))
         .select("id", 
                 F.col("new.names").alias("names"), 
                 F.col("new.score").alias("score")
                )

我得到一个错误：

TypeError: zip argument #1 must support iteration

我还尝试使用rdd flatMap()进行爆炸，但仍然得到相同的错误

有没有其他方法可以实现这一点

Answer 1

使用^{}函数+explode，然后最后使用group by和pivot：

from pyspark.sql.functions import explode, expr, first

df1 = df.withColumn("name_score", explode(expr("arrays_zip(name, score)"))) \
       .selectExpr("id", "name_score.*")\
       .groupBy("id").pivot("name").agg(first("score"))

df1.show(truncate=False)

#+  +  -+   -+   +   + -+  +
#|id  |F1   |F2     |F3    |F4    |F5 |F6  |
#+  +  -+   -+   +   + -+  +
#|ab01|00123|0.001  |0.127 |0.0123|111|null|
#|ab04|00345|0.01112|0.1567|0.0186|555|null|
#|ab03|00234|0.078  |0.188 |0.0144|188|null|
#|ab02|00124|0.003  |0.156 |0.067 |156|254 |
#+  +  -+   -+   +   + -+  +

对于Spark<；2.4：

您的UDFcombine正常。出现错误TypeError: zip argument #1 must support iteration，因为要将列名作为字符串传递给UDF，请使用col传递列：

df1 = df.withColumn("name_score", explode(combine(col("name"), col("score")))) \
      .selectExpr("id", "name_score.*") \
      .groupBy("id").pivot("name").agg(first("score"))