300万行Polars DataFrame转PySpark DataFrame的高效方法求助

核心问题背景 #

手里有个300万行×145列的超大Polars DataFrame，要转成PySpark DataFrame，试过转Pandas（不管开不开Arrow扩展都有问题：要么类型不兼容，要么内存爆了）、转字典也OOM，还遇到Spark Schema验证的奇怪问题。

推荐解决方案 #

方法1：Polars + PyArrow分批次导入（内存友好+高效） #

利用Polars原生支持Arrow格式的特性，分批次生成Arrow数据块，直接喂给Spark，全程不用转Pandas或字典，从根源避免OOM：

import polars as pl
import pyarrow as pa

# 批次大小根据内存情况调，比如10万行一批，内存够可以往大调
BATCH_SIZE = 100000

# 生成Arrow批次迭代器，每次只加载部分数据到内存
def polars_arrow_batches(df: pl.DataFrame, batch_size: int):
    for start in range(0, len(df), batch_size):
        yield df.slice(start, batch_size).to_arrow()

# 直接用Arrow批次创建Spark DataFrame
spark_df = spark.createDataFrame(
    polars_arrow_batches(polars_dataframe, BATCH_SIZE),
    schema=my_spark_schema
)

优势

• 无中间冗余数据，类型转换更精准
• 分批次加载，完全避免一次性占满内存
• Spark对Arrow格式原生支持，性能比转Pandas快很多

方法2：Polars直接写入Spark（0.20+版本适用） #

如果你的Polars版本在0.20及以上，可以直接把Polars DataFrame写入Spark表，全程绕开内存中转，适合超大数据量：

# 直接写入目标Spark表，支持覆盖、分区等配置
polars_dataframe.write_spark(
    table="your_db.your_table",
    mode="overwrite",
    partition_by=["your_partition_col"]  # 可选，按列分区优化查询
)

# 如果需要内存中的Spark DataFrame，先写临时表再读取
polars_dataframe.write_spark(table="temp.temp_polars_data", mode="overwrite")
spark_df = spark.table("temp.temp_polars_data")

关于你遇到的Schema验证奇怪现象 #

你发现的verifySchema=True失败但最终DataFrame一致的问题，本质是Spark的Schema校验逻辑没跟上PyArrow扩展类型：

• Polars用PyArrow扩展类型存可空值（比如NAType），但Spark的校验逻辑只认标准Python类型（像datetime、np.nan）
• 但实际写入时，Spark会自动把这些Arrow扩展类型转成兼容的Spark类型，所以最终结果完全一致
• 用上面的Arrow批次方法，Spark能直接识别Arrow类型，不需要关闭verifySchema

额外优化提示 #

• 调优批次大小：内存充足就把BATCH_SIZE设大（比如20万行），内存吃紧就调小（比如5万行）
• 提前对齐类型：在Polars阶段就把类型调成Spark兼容的，比如把datetime[ns]转成带时区的datetime[ns, UTC]，避免时区转换问题；统一空值格式
• 开启Spark Arrow优化：如果偶尔需要用Pandas中转，设置spark.sql.execution.arrow.pyspark.enabled=true能提升转换速度，但大内存场景还是优先分批次

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