如何在 PySpark 中动态创建包含命名字段的结构体数组-Python教程-PHP中文网

如何在 PySpark 中动态创建包含命名字段的结构体数组

本文介绍如何使用 PySpark 的 SQL 函数（如 TRANSFORM、named_struct 和 ARRAY）将指定列名列表动态转换为一个结构体数组，每个结构体包含 'key'（列名）和 'value'（对应列值）两个字段，避免硬编码并支持任意列组合。

本文介绍如何使用 pyspark 的 sql 函数（如 `transform`、`named_struct` 和 `array`）将指定列名列表动态转换为一个结构体数组，每个结构体包含 `'key'`（列名）和 `'value'`（对应列值）两个字段，避免硬编码并支持任意列组合。

在 PySpark 中，若需将多列动态构造成 array> 形式的列，核心难点在于：SQL 表达式中的 TRANSFORM 无法直接访问列名字符串（如 'name'），只能访问列值（如 col）；而 named_struct 的字段名必须是字面量或表达式结果，不能直接插值列标识符。

你尝试的 f"named_struct('key', {col}, 'val', col)" 失败，是因为 Python 的 f-string 在 SQL 字符串生成阶段就求值了，而 col 是 SQL 执行时的变量，此时 Python 环境中根本不存在该变量 —— 这属于典型的“运行时 vs 编译时”混淆。

✅ 正确解法是：使用 arrays_zip + transform 组合，先将列名数组与列值数组配对，再逐对构造结构体。这是 Spark 3.0+ 推荐的动态键值映射模式。

✅ 推荐实现（Spark 3.0+）

from pyspark.sql import Row
from pyspark.sql.functions import arrays_zip, transform, col, array, lit, named_struct
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 构建示例 DataFrame
data = [
    Row(name='Alice', age=30, city='New York'),
    Row(name='Bob', age=25, city='Los Angeles'),
    Row(name='Charlie', age=35, city='Chicago')
]
df = spark.createDataFrame(data)

# 指定要纳入结构体的列名（可动态传入）
cols = ['name', 'age']

# 步骤分解：
# 1. 构造列名数组（字面量）
keys_array = array([lit(c) for c in cols])
# 2. 构造对应列值数组
values_array = array([col(c) for c in cols])
# 3. zip 列名与列值 → array<struct ... col2:>>
zipped = arrays_zip(keys_array, values_array)
# 4. transform 每个 struct，重命名为 key/value
result_structs = transform(
    zipped,
    lambda x: named_struct(
        lit('key'), x['keys_array'],   # 注意：zip 后字段名为原数组名（默认为 'col1', 'col2'...）
        lit('value'), x['values_array']
    )
)

df_with_structs = df.withColumn('array_of_structs', result_structs)
df_with_structs.show(truncate=False)</struct>

输出结果：

+-------+---+-----------+--------------------------+
|name   |age|city       |array_of_structs          |
+-------+---+-----------+--------------------------+
|Alice  |30 |New York   |[{name, Alice}, {age, 30}]|
|Bob    |25 |Los Angeles|[{name, Bob}, {age, 25}]  |
|Charlie|35 |Chicago    |[{name, Charlie}, {age, 35}]|
+-------+---+-----------+--------------------------+

⚠️ 注意事项

arrays_zip 要求输入数组长度一致，确保 keys_array 与 values_array 元素数量相同；
Spark 3.4+ 中 arrays_zip 默认字段名为 col1, col2…，因此 x['keys_array'] 实际应写为 x['col1'] —— 但更健壮写法是显式别名（见下方优化版）；
若列类型不一致（如 string + int），named_struct 会自动推断 value 字段为 uniontype，建议统一 cast 或接受其灵活性。

? 健壮增强版（兼容 Spark 3.1+，显式字段名）

from pyspark.sql.functions import expr

# 使用 SQL 表达式显式命名 zip 字段（避免依赖默认名）
zipped_named = expr(f"arrays_zip({', '.join([f'array({repr(c)})' for c in cols])}, {', '.join([f'array({c})' for c in cols])})")
# 然后 transform（注意字段引用改为 'col1', 'col2'）
result_structs_safe = transform(
    zipped_named,
    lambda x: named_struct(lit('key'), x['col1'], lit('value'), x['col2'])
)