带VariantType列的PySpark DataFrame转Pandas触发'NoneType不可迭代'错误

问题说明 #

将包含VariantType列的PySpark DataFrame转换为Pandas DataFrame时，调用toPandas()会抛出'NoneType' object is not iterable错误，无论是自定义代码还是Spark官方文档中的示例代码都会出现该问题。

错误原因 #

这是PySpark的已知bug：在处理Variant类型到Pandas的转换时，当内部触发MALFORMED_VARIANT错误断言时，messageParameters参数被传递为None，导致后续代码尝试对None执行set()操作，触发迭代类型错误。该bug存在于PySpark 3.4.x、3.5.x的部分版本中。

解决方案 #

方案1：升级PySpark到修复版本 #

升级PySpark至3.4.3+或3.5.1+版本（具体版本以官方修复记录为准），官方已在这些版本中修复了该错误处理逻辑的问题。

方案2：转换前兼容Variant列类型 #

如果暂时无法升级PySpark，可以先将Variant列转换为Pandas兼容的类型，再执行转换：

方法A：将Variant列转为JSON字符串

通过to_json()函数把Variant列转为JSON格式字符串，避免直接转换Variant类型：

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr, to_json

spark = SparkSession.builder.getOrCreate()
data = [
    ('{"name": "Alice", "age": 30, "city": "New York", "salary": 85000}',),
    ('{"name": "Bob", "age": 25, "city": "San Francisco", "salary": 92000}',),
    ('{"name": "Charlie", "age": 35, "city": "Los Angeles", "salary": 79000}',)
]
df = spark.createDataFrame(data, ["json_data"])
df_variant = df.withColumn("variant_column", expr("parse_json(json_data)"))

# 将Variant列转为JSON字符串，兼容Pandas转换
df_compatible = df_variant.withColumn("variant_column", to_json("variant_column"))
pdf = df_compatible.toPandas()
print(pdf)

方法B：展开Variant列为结构化列

如果Variant列的JSON结构固定，可以直接解析并展开为单独的DataFrame列，完全规避Variant类型：

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr, col

spark = SparkSession.builder.getOrCreate()
data = [
    ('{"name": "Alice", "age": 30, "city": "New York", "salary": 85000}',),
    ('{"name": "Bob", "age": 25, "city": "San Francisco", "salary": 92000}',),
    ('{"name": "Charlie", "age": 35, "city": "Los Angeles", "salary": 79000}',)
]
df = spark.createDataFrame(data, ["json_data"])
df_variant = df.withColumn("variant_column", expr("parse_json(json_data)"))

# 展开Variant列的所有字段为单独列
df_expanded = df_variant.select(
    "json_data",
    col("variant_column.name").alias("name"),
    col("variant_column.age").alias("age"),
    col("variant_column.city").alias("city"),
    col("variant_column.salary").alias("salary")
)
pdf = df_expanded.toPandas()
print(pdf)

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