PySpark启用Arrow优化转换结构体数组报错，求解决方案

问题场景 #

已在Spark会话中启用Arrow优化参数：

app_name = "App"

spark_conf = {
    # 其他参数
    'spark.sql.execution.arrow.enabled': 'true'
}

builder = (
    SparkSession
    .builder
    .appName(app_name)
)

for k, v in spark_conf.items():
    builder.config(k, v)
    
spark = builder.getOrCreate()

当对包含嵌套结构体数组的DataFrame调用toPandas时：

df.printSchema()
"""
root
 |-- x: string (nullable = true)
 |-- dtm: long (nullable = true)
 |-- items: array (nullable = false)
 |    |-- element: struct (containsNull = false)
 |    |    |-- campaign_id: long (nullable = true)
 |    |    |-- cr_pred: float (nullable = true)
 |    |    |-- ctr_pred: float (nullable = true)
"""

df.toPandas()

触发如下错误：

toPandas attempted Arrow optimization because 'spark.sql.execution.arrow.pyspark.enabled' is set to true; however, failed by the reason below:
  Unsupported type in conversion to Arrow: ArrayType(StructType([StructField('campaign_id', LongType(), True), StructField('cr_pred', FloatType(), True), StructField('ctr_pred', FloatType(), True)]), True)

原因说明 #

虽然Arrow原生支持结构体数组类型，但PySpark与Arrow的转换逻辑是分版本迭代完善的。早期Spark版本（3.0以下）的Arrow转换层未实现对数组嵌套结构体这类复杂类型的支持，导致启用Arrow优化时触发转换失败。

解决方案 #

1. 升级Spark版本（推荐） #

Spark 3.0及以上版本已完善了对嵌套类型的Arrow转换支持，升级后无需额外修改代码，直接使用toPandas即可触发Arrow优化完成转换。

2. 临时处理嵌套列（无法升级版本时） #

如果无法升级Spark，可以将嵌套列转换为JSON字符串，保留其他列的Arrow优化，待转换为Pandas DataFrame后再解析JSON：

from pyspark.sql.functions import to_json, col

# 将嵌套结构体数组转为JSON字符串
df_processed = df.withColumn("items", to_json(col("items")))

# 启用Arrow转换生成Pandas DataFrame
pd_df = df_processed.toPandas()

# 在Pandas中解析JSON列
import pandas as pd
pd_df["items"] = pd_df["items"].apply(lambda x: pd.read_json(x, orient="records"))

3. 临时禁用Arrow转换（兜底方案） #

如果上述方法都不可行，可以临时关闭当前会话的Arrow优化，代价是数据传输速度会回落至原生Spark-Python转换水平：

# 临时禁用Arrow优化
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "false")

# 转换为Pandas DataFrame
pd_df = df.toPandas()

# 恢复Arrow优化（按需执行）
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

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