如何将PySpark DataFrame指定列数据转换为struct结构体数组

方案1：使用内置函数实现（推荐，性能最优） #

适用于大规模数据处理，无Python UDF序列化开销，转换效率更高。
前置假设：你已拆分得到的字符串数组列名为value_arr，目标struct默认包含key（存储id1/id2标识）、val（存储对应值）两个字段，可按需调整字段名。

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

# 定义目标struct数组的schema
target_schema = ArrayType(StructType([
    StructField("key", StringType(), nullable=True),
    StructField("val", StringType(), nullable=True)
]))

# 执行转换，split的第三个参数限制只拆分第一个冒号，避免value中包含冒号时拆分错误
df = df.withColumn(
    "struct_arr",
    F.expr("transform(value_arr, item -> struct(split(item, ':', 2)[0] as key, split(item, ':', 2)[1] as val))")
)

方案2：UDF实现（适合复杂自定义逻辑） #

如果需要增加格式校验、特殊值处理等复杂逻辑，可以使用Python UDF实现：

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

# 定义目标struct数组的schema
target_schema = ArrayType(StructType([
    StructField("key", StringType(), nullable=True),
    StructField("val", StringType(), nullable=True)
]))

# 自定义转换逻辑
def parse_arr_to_struct(arr):
    result = []
    for item in arr:
        if not item or ":" not in item:
            # 异常格式处理逻辑可按需调整，比如填充空值、跳过或抛出异常
            continue
        key, val = item.split(":", 1)
        result.append({"key": key, "val": val})
    return result

# 注册UDF并执行转换
struct_convert_udf = F.udf(parse_arr_to_struct, target_schema)
df = df.withColumn("struct_arr", struct_convert_udf(F.col("value_arr")))

结果验证 #

# 查看新列的schema结构是否符合预期
df.printSchema()
# 查看转换后的实际数据
df.select("struct_arr").show(truncate=False)

注意事项 #

• 可根据业务需求修改struct的字段名、字段类型，比如值为纯数字时可将val的类型改为IntegerType
• 数据量较大时优先使用内置函数方案，性能比Python UDF高10倍以上

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