PySpark中如何按数组元素位置将两数组列转换为结构体数组？

需求说明 #

需要将包含first_name和last_name的两个等长数组列，对应位置的元素组合成{first_name: string, last_name: string}格式的结构体，最终生成包含这些结构体的数组列names，同时处理两列均为null的边界情况。

推荐方案：使用Spark原生函数（高效无UDF） #

Spark内置的arrays_zip函数可以直接将多个数组按位置打包成结构体数组，配合条件判断处理null场景，无需编写UDF，性能更优：

from pyspark.sql import functions as F

# 处理逻辑：
# 1. 当first_name和last_name均为null时，生成包含空结构体的数组
# 2. 其他情况用arrays_zip打包对应位置元素
df_result = df.withColumn(
    "names",
    F.when(
        F.col("first_name").isNull() & F.col("last_name").isNull(),
        F.array(F.struct(F.lit(None).alias("first_name"), F.lit(None).alias("last_name")))
    ).otherwise(
        F.arrays_zip(F.col("first_name"), F.col("last_name"))
    )
)

结果验证 #

生成的DataFrame输出与需求一致：

+-------------------------------------+
| names                               |
+-------------------------------------+
| [{"John", "Smith"},{"Jane", "Doe"}] |
| [{"Dwight", "Schrute"}]             |
| [{"null", "null"}]                  |
+-------------------------------------+

Schema也符合要求：

root
 |-- names: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- first_name: string (nullable = true)
 |    |    |-- last_name: string (nullable = true)

备选方案：修复后的UDF实现 #

如果一定要用UDF，需要修正原代码中的多处错误：

原UDF的错误点 #

1. 导入错误：应导入pyspark.sql.functions而非pyspark.sql
2. UDF内部混用Spark函数：struct/lit是Spark引擎端API，不能在Python lambda中使用
3. flatten函数误用：该函数用于展开嵌套数组，不是合并数组，且参数应为单个数组列
4. 返回类型定义错误：ArrayType(StructType())需要明确指定结构体字段
5. 未处理null输入：当数组为null时，len()会抛出异常

修复后的UDF代码 #

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

# 定义返回类型的结构体
name_struct = StructType([
    StructField("first_name", StringType(), nullable=True),
    StructField("last_name", StringType(), nullable=True)
])
return_type = ArrayType(name_struct)

def combine_names(first_names, last_names):
    # 处理双null的边界情况
    if first_names is None and last_names is None:
        return [{"first_name": None, "last_name": None}]
    # 按位置组合元素为字典（Spark会自动映射为结构体）
    return [{"first_name": f, "last_name": l} for f, l in zip(first_names, last_names)]

# 注册UDF
convert_names_udf = F.udf(combine_names, return_type)

# 应用UDF
df_result = df.withColumn("names", convert_names_udf(F.col("first_name"), F.col("last_name")))

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