使用pyarrow.float64()创建PySpark DataFrame遇DoubleScalar Schema推断错误

错误原因 #

直接将PyArrow Array传入createDataFrame时，Spark会尝试将数组中的每个Scalar元素当作独立行，但无法自动推断PyArrow Scalar的Schema，因此触发CANNOT_INFER_SCHEMA_FOR_TYPE错误。

解决方案（保留PyArrow类型） #

方法1：通过PyArrow Table创建DataFrame #

Spark原生支持从PyArrow Table读取数据，能完整保留原类型映射。将PyArrow Array包装为Table后，再传入createDataFrame即可：

import pyarrow
import pyspark

session = pyspark.sql.SparkSession.builder \
                              .appName ("pyspark_pyarrow") \
                              .config ("spark.executor.memory", "2g") \
                              .config ("spark.sql.shuffle.partitions", "2") \
                              .config ("spark.driver.host", "localhost") \
                              .config ("spark.sql.execution.arrow.pyspark.enabled", "true") \
                              .getOrCreate()

session.sparkContext.setLogLevel ("DEBUG")

# 创建PyArrow float64类型数组
array = pyarrow.array([1, 2, 3, 4, 5, 6, 7, 8, 9], type=pyarrow.float64())

# 转换为PyArrow Table，指定列名
pa_table = pyarrow.Table.from_arrays([array], names=["value"])

# 基于Table创建DataFrame
df = session.createDataFrame(pa_table)

# 验证类型与数据
df.printSchema()
df.show()

session.stop ()

方法2：手动指定Spark Schema #

手动定义与PyArrow float64对应的Spark DoubleType Schema，同时将PyArrow数组转为Spark可识别的行结构：

import pyarrow
import pyspark
from pyspark.sql.types import StructType, StructField, DoubleType

session = pyspark.sql.SparkSession.builder \
                              .appName ("pyspark_pyarrow") \
                              .config ("spark.executor.memory", "2g") \
                              .config ("spark.sql.shuffle.partitions", "2") \
                              .config ("spark.driver.host", "localhost") \
                              .config ("spark.sql.execution.arrow.pyspark.enabled", "true") \
                              .getOrCreate()

session.sparkContext.setLogLevel ("DEBUG")

array = pyarrow.array([1, 2, 3, 4, 5, 6, 7, 8, 9], type=pyarrow.float64())

# 定义匹配float64的Schema
schema = StructType([StructField("value", DoubleType(), nullable=True)])

# 将PyArrow数组转为行数据集合
data = [(val,) for val in array.to_pylist()]

# 创建DataFrame时指定Schema
df = session.createDataFrame(data, schema=schema)

# 验证类型与数据
df.printSchema()
df.show()

session.stop ()

两种方法都能确保全程使用float64（对应Spark DoubleType）类型，且保留PyArrow的类型特性。

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