PySpark动态生成StructType：如何自动转换values列为统一StructType？

现有如下数据表，其中values和schema列均以StringType存储：

需求目标 #

希望将values列转换为包含所有可能键的StructType，最终Schema如下：

|-- id: integer (nullable = false)
 |-- values: struct (nullable = true)
 |    |-- colA: double (nullable = true)
 |    |-- colB: string (nullable = true)
 |    |-- colC: double (nullable = true)
 |    |-- colG: string (nullable = true)
 |    |-- colH: double (nullable = true)
 |    |-- colJ: double (nullable = true)
 |    |-- colM: string (nullable = true)
 |    |-- colX: double (nullable = true)

核心问题 #

能否无需显式指定上述Schema即可实现转换？实际场景涉及数百个字段，理想情况是不依赖原schema列自动推断Schema，使用该列也可接受（注意schema列中的数据类型不一定匹配Spark数据类型）。数据需存储为Delta格式，是否有可行方案，还是只能存储为MapType(StringType, StringType)？

测试DataFrame创建代码 #

import pyspark.sql.types as T

data = [
    (
        2,
        "{'colA':3.2, 'colB':'val2', 'colC':3.4}",
        "{'colA':'FLOAT', 'colB':'STRING', 'colC':'FLOAT'}",
    ),
    (
        3, 
        "{'colC':3.2, 'colX':3.9}", 
        "{'colC':'FLOAT', 'colX':'FLOAT'}"),
    (
        4, 
        "{'colG':'val1', 'colH':93.2}", 
        "{'colG':'STRING', 'colH':'FLOAT'}"),
    (
        5,
        "{'colG':'val4', 'colA':4.2, 'colJ':93.2, 'colM':'val4'}",
        "{'colG':'STRING', 'colA':'FLOAT', 'ColJ':'FLOAT', 'ColM':'STRING'}",
    ),
]

schema = T.StructType(
    [
        T.StructField("id", T.IntegerType()),
        T.StructField("values", T.StringType()),
        T.StructField("schema", T.StringType()),
    ]
)

df = spark.createDataFrame(data, schema)

完全可以不用显式指定Schema实现转换，下面提供两种可行方案：

方案一：不依赖原schema列，自动推断全量字段类型 #

步骤说明 #

1. 将values字符串解析为Map类型，展开临时字段
2. 收集所有出现过的字段名，统一推断每个字段的全局兼容类型
3. 按推断类型提取字段并组合成Struct

代码实现 #

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

# 1. 解析values为Map类型
df_parsed = df.withColumn("values_map", F.from_json("values", F.schema_of_json(df.select("values").first()[0])))

# 2. 收集所有字段名
all_fields = set()
for row in df_parsed.select(F.map_keys("values_map")).collect():
    all_fields.update(row[0])
all_fields = sorted(all_fields)

# 3. 推断每个字段的全局兼容类型
field_types = {}
for field in all_fields:
    # 取第一个非空值推断类型，可根据数据规模调整为采样更多值
    sample_val = df_parsed.filter(F.col(f"values_map.{field}").isNotNull()).select(F.col(f"values_map.{field}")).first()
    if sample_val:
        field_types[field] = sample_val[0].__class__.__name__
    else:
        field_types[field] = "StringType"  # 空字段默认字符串类型

# 4. 构建StructType并转换
struct_fields = []
for field in all_fields:
    spark_type = getattr(T, field_types[field])()
    struct_fields.append(StructField(field, spark_type, nullable=True))
target_struct = StructType(struct_fields)

# 将Map转换为Struct
result_df = df_parsed.withColumn("values", F.struct(*[F.col(f"values_map.{f}").alias(f) for f in all_fields])).drop("values_map", "schema")

# 查看最终Schema
result_df.printSchema()

方案二：利用原schema列生成全局Struct #

步骤说明 #

1. 解析所有schema列的字符串为Map，收集全量字段及其声明类型
2. 将声明类型映射为Spark支持的数据类型
3. 构建全局StructType，直接解析values为该Struct

代码实现 #

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

# 类型映射表，根据实际场景补充更多类型
type_mapping = {
    "FLOAT": T.DoubleType(),
    "STRING": T.StringType(),
    "INT": T.IntegerType(),
    "BOOLEAN": T.BooleanType()
}

# 1. 解析所有schema列，收集全量字段和类型（统一字段名大小写）
all_schema_fields = {}
for row in df.select(F.from_json("schema", F.schema_of_json(df.select("schema").first()[0]))).collect():
    schema_map = row[0].asDict()
    for k, v in schema_map.items():
        normalized_key = k.lower()
        if normalized_key not in all_schema_fields:
            all_schema_fields[normalized_key] = v

# 2. 构建全局StructType
struct_fields = []
for field_name, field_type in sorted(all_schema_fields.items()):
    spark_type = type_mapping.get(field_type.upper(), T.StringType())  # 未知类型默认字符串
    struct_fields.append(StructField(field_name, spark_type, nullable=True))
target_struct = StructType(struct_fields)

# 3. 解析values为目标Struct
result_df = df.withColumn("values", F.from_json("values", target_struct)).drop("schema")

# 查看最终Schema
result_df.printSchema()

存储为Delta格式 #

两种方案生成的DataFrame都可直接存储为Delta格式：

result_df.write.format("delta").mode("overwrite").save("/path/to/delta_table")

关键说明 #

• 方案一的自动推断需确保采样值能代表字段真实类型，若存在类型冲突（同一字段既有数值又有字符串），会以首次采样类型为准，可根据数据调整推断逻辑
• 方案二需维护类型映射表，处理原schema列与Spark类型的差异，同时注意字段名大小写不一致的问题（示例中已做统一小写处理）
• 两种方案都避免了手动编写数百个字段的Schema，无需退化为MapType存储

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