如何将空Pandas DataFrame安全转换为PySpark DataFrame并保留数据类型？

当Pandas DataFrame为空时，spark.createDataFrame会因无法推断schema报错ValueError: can not infer schema from empty dataset，以下是两种简便的解决方案：

方法一：临时填充一行数据（最简单通用） #

通过临时插入一行匹配列类型的默认数据，让Spark能正确推断schema，再通过limit(0)得到空的Spark DataFrame，无需手动处理复杂类型：

import pandas as pd
from pyspark.sql import SparkSession

# 假设Spark Session已初始化
spark = SparkSession.builder.appName("EmptyDFConversion").getOrCreate()

raw_data = []
cols = ['col_1', 'col_2', 'col_3']
types_dict = {
    'col_1': str,
    'col_2': float,
    'col_3': bool
}
pandas_df = pd.DataFrame(raw_data, columns=cols).astype(types_dict)

if pandas_df.empty:
    # 构造匹配类型的临时默认行
    default_row = {
        'col_1': '',
        'col_2': 0.0,
        'col_3': False
    }
    temp_pandas_df = pd.DataFrame([default_row])
    # 生成Spark DF后取空结果
    spark_df = spark.createDataFrame(temp_pandas_df).limit(0)
else:
    spark_df = spark.createDataFrame(pandas_df)

# 验证schema
spark_df.printSchema()

该方法优势是兼容性强，自动适配包括结构体、日期在内的复杂类型，操作简单高效。

方法二：自动生成Spark Schema（精准控制场景） #

基于Pandas DataFrame的dtypes自动映射生成Spark Schema，无需临时数据，适合需要精确控制字段属性的场景：

from pyspark.sql.types import StructType, StructField, StringType, FloatType, BooleanType
import pandas as pd
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("EmptyDFConversion").getOrCreate()

raw_data = []
cols = ['col_1', 'col_2', 'col_3']
types_dict = {
    'col_1': str,
    'col_2': float,
    'col_3': bool
}
pandas_df = pd.DataFrame(raw_data, columns=cols).astype(types_dict)

def pandas_type_to_spark_type(pd_type):
    # 基础类型映射
    if pd_type == object:
        return StringType()
    elif pd_type == float:
        return FloatType()
    elif pd_type == bool:
        return BooleanType()
    # 日期时间类型
    elif pd_type == 'datetime64[ns]':
        from pyspark.sql.types import TimestampType
        return TimestampType()
    # 嵌套Map类型（Pandas dict）
    elif hasattr(pd_type, '__origin__') and pd_type.__origin__ == dict:
        from pyspark.sql.types import MapType
        key_type = pandas_type_to_spark_type(pd_type.__args__[0])
        value_type = pandas_type_to_spark_type(pd_type.__args__[1])
        return MapType(key_type, value_type)
    # 嵌套Array类型（Pandas list）
    elif hasattr(pd_type, '__origin__') and pd_type.__origin__ == list:
        from pyspark.sql.types import ArrayType
        element_type = pandas_type_to_spark_type(pd_type.__args__[0])
        return ArrayType(element_type)
    # 可按需扩展其他类型
    else:
        raise ValueError(f"不支持的Pandas类型: {pd_type}")

# 构建Spark Schema
schema = StructType([
    StructField(col, pandas_type_to_spark_type(pandas_df.dtypes[col]), nullable=True)
    for col in pandas_df.columns
])

# 创建空Spark DataFrame
spark_df = spark.createDataFrame(pandas_df, schema=schema)

# 验证schema
spark_df.printSchema()

这个方法通过类型映射函数自动处理嵌套结构，避免手动编写繁琐的schema定义。

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