在PySpark DataFrame单列存储不同Schema的实现问询

直接把StructType对象放进DataRow创建DataFrame会抛出ValueError: Some of types cannot be determined after inferring，核心原因是Spark无法自动推断这种复杂自定义类型的Schema。但这个需求是可以实现的，推荐两种实用方案：

方案一：将Schema序列化为JSON字符串存储 #

这是最简便且易维护的方式，StructType本身支持和JSON格式互相转换，后续使用时再反序列化即可。

示例代码：

import pyspark.sql.functions as F
import json
from pyspark.sql.types import *

# 把StructType转成JSON字符串存入列表
A = [
    {"TableName": "Table1", "Schema": StructType([StructField("a", StringType()), StructField("b", IntegerType())]).json()}
    , {"TableName": "Table2", "Schema": StructType([StructField("b", StringType()), StructField("c", IntegerType())]).json()}
]

# 创建DataFrame，此时Schema列是字符串类型
df_A = spark.createDataFrame(A)
df_A.show(truncate=False)

# 后续需要使用Schema时，反序列化回StructType
def get_schema_from_json(json_str):
    return StructType.fromJson(json.loads(json_str))

# 如果需要在DataFrame中批量处理，可以注册UDF
schema_udf = F.udf(get_schema_from_json, StructType())

方案二：用BinaryType存储序列化后的Schema对象 #

如果需要保留对象的二进制形态，可以用pickle序列化后存储为二进制类型，但这种方式不跨语言、可读性差，仅适合特定场景。

示例代码：

import pickle
import pyspark.sql.functions as F
from pyspark.sql.types import *

# 序列化StructType为二进制数据
def serialize_schema(schema):
    return pickle.dumps(schema)

A = [
    {"TableName": "Table1", "Schema": serialize_schema(StructType([StructField("a", StringType()), StructField("b", IntegerType())]))}
    , {"TableName": "Table2", "Schema": serialize_schema(StructType([StructField("b", StringType()), StructField("c", IntegerType())]))}
]

# 手动指定DataFrame的Schema
df_schema = StructType([
    StructField("TableName", StringType()),
    StructField("Schema", BinaryType())
])

df_A = spark.createDataFrame(A, schema=df_schema)

# 反序列化使用Schema
def deserialize_schema(bin_data):
    return pickle.loads(bin_data)

优先推荐第一种方案，JSON格式可读、跨语言，方便后续调试和维护。

内容的提问来源于stack exchange，提问作者Quynh-Mai Chu