Pyspark重排结构体数组为含数组成员的结构体报错求解

你写的UDF里错误使用了Spark内置函数F.array、F.struct：UDF执行在Worker节点的Python进程中，输入的arr和内部迭代的x都是Python原生的列表、字典对象，不是Spark的Column对象，不能调用Pyspark的sql函数接口，才会触发_jvm相关的空指针报错。

方法1：修正UDF为纯Python逻辑 #

把UDF内部的逻辑换成Python原生的列表、字典构造，不需要调用Pyspark函数：

from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StructType, StructField, StringType

concat_udf = F.udf(
    lambda arr: [
        {"nome": x["nome"], "circadiana": [x[f"h_{i}"] for i in range(24)]} 
        for x in arr
    ],
    ArrayType(
        StructType([
            StructField("nome", StringType(), True),
            StructField("circadiana", ArrayType(StringType()), True)
        ])
    )
)

# 调用方式不变
df_new = df.withColumn("area_provenienza_X", concat_udf(F.col("array_area")))
df_new.show()

方法2：用Spark内置函数实现（性能更优，避免UDF序列化开销） #

对于数组操作可以用transform内置函数直接处理，不需要自定义UDF，执行效率更高：

df_new = df.withColumn(
    "area_provenienza_X",
    F.transform(
        "array_area",
        lambda x: F.struct(
            x["nome"].alias("nome"),
            F.array(*[x[f"h_{i}"] for i in range(24)]).alias("circadiana")
        )
    )
)
df_new.show()

内容的提问来源于stack exchange，提问作者Catanzaro