如何将PySpark Pandas DataFrame转换为PySpark SQL DataFrame？

1. 转换为PySpark SQL DataFrame #

PySpark Pandas DataFrame提供了原生的to_spark()方法，可直接将其转换为PySpark SQL DataFrame：

# 假设你的PySpark Pandas DataFrame变量名为ps_df
spark_df = ps_df.to_spark()

2. 强制Schema写入Delta #

转换为Spark原生DataFrame后，即可使用Spark的Write API指定目标Schema，实现schema-on-write的强制约束。步骤如下：

• 先定义你需要的目标Schema（使用StructType）
• 调用Spark Write API时通过.schema()传入该Schema

示例代码：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

# 定义要强制的目标Schema
target_schema = StructType([
    StructField("user_id", IntegerType(), nullable=False),
    StructField("username", StringType(), nullable=True),
    StructField("score", DoubleType(), nullable=True),
    StructField("register_date", StringType(), nullable=True)
])

# 写入Delta文件，强制使用指定Schema
spark_df.write \
    .format("delta") \
    .schema(target_schema) \
    .mode("overwrite")  # 根据业务需求选择模式：overwrite/append/ignore等
    .save("/your/delta/storage/path")

关键说明 #

• PySpark Pandas的to_delta方法确实未支持直接传入Schema参数，转成Spark原生DataFrame是最直接的解决方案
• 指定Schema后，Spark会在写入时严格按照定义的结构和类型处理数据：若原数据类型与目标Schema兼容则自动转换，不兼容则抛出异常（可通过Spark配置调整容错行为）
• 如果需要创建Delta表而非仅保存文件，可将.save()替换为.saveAsTable("database.table_name")，同样支持.schema()参数

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