如何将已存在的Spark DataFrame的Schema修改为指定自定义Schema？

有两种实用方法可以实现这个需求：

方法一：逐个字段强制类型转换（适合字段数量较少的场景） #

针对自定义Schema中的每个字段，将原DataFrame对应的字段强制转换为目标类型，再重新组装成新的DataFrame：

from pyspark.sql.types import *

# 定义你提供的自定义Schema
custom_schema = StructType([
    StructField("_links", MapType(StringType(), MapType(StringType(), StringType()))),
    StructField("identifier", StringType()),
    StructField("enabled", BooleanType()),
    StructField("family", StringType()),
    StructField("categories", ArrayType(StringType())),
    StructField("groups", ArrayType(StringType())),
    StructField("parent", StringType()),
    StructField("values", MapType(StringType(), ArrayType(MapType(StringType(), StringType())))),
    StructField("created", StringType()),
    StructField("updated", StringType()),
    StructField("associations", MapType(StringType(), MapType(StringType(), ArrayType(StringType())))),
    StructField("quantified_associations", MapType(StringType(), IntegerType())),
    StructField("metadata", MapType(StringType(), StringType()))
])

# 假设原DataFrame名为df
df_modified = df.select(
    df["_links"].cast(MapType(StringType(), MapType(StringType(), StringType()))).alias("_links"),
    df["identifier"].cast(StringType()).alias("identifier"),
    df["enabled"].cast(BooleanType()).alias("enabled"),
    df["family"].cast(StringType()).alias("family"),
    df["categories"].cast(ArrayType(StringType())).alias("categories"),
    df["groups"].cast(ArrayType(StringType())).alias("groups"),
    df["parent"].cast(StringType()).alias("parent"),
    df["values"].cast(MapType(StringType(), ArrayType(MapType(StringType(), StringType())))).alias("values"),
    df["created"].cast(StringType()).alias("created"),
    df["updated"].cast(StringType()).alias("updated"),
    df["associations"].cast(MapType(StringType(), MapType(StringType(), ArrayType(StringType())))).alias("associations"),
    df["quantified_associations"].cast(MapType(StringType(), IntegerType())).alias("quantified_associations"),
    df["metadata"].cast(MapType(StringType(), StringType())).alias("metadata")
)

# 验证转换后的Schema是否匹配
df_modified.printSchema()

方法二：转JSON字符串后重新读取（适合字段数量较多的场景） #

将原DataFrame转换为JSON格式的RDD，再使用自定义Schema重新读取，一次性应用整个Schema规则：

# 原DataFrame转为JSON字符串RDD
json_rdd = df.toJSON()

# 用自定义Schema重新加载数据
df_modified = spark.read.schema(custom_schema).json(json_rdd)

# 验证Schema
df_modified.printSchema()

注意事项 #

• 转换时要确保原字段内容与目标类型兼容，比如enabled字段如果是字符串"true"/"false"，转BooleanType可以正常转换；若为不兼容值，会生成null或触发报错。
• 若原DataFrame存在自定义Schema中没有的字段，转换后这些字段会被丢弃；若自定义Schema包含原DataFrame没有的字段，会新增该字段并赋值为null。

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