Databricks中Spark读取ZIP解压后SPSS .sav文件的报错与性能问题

1. 解决read_sav()参数错误问题 #

报错read_sav() got an unexpected keyword argument 'apply_value_labels'，核心原因是混淆了pyreadstat和pandas的SPSS读取函数：

• pyreadstat.read_sav()支持apply_value_labels参数，用于将变量标签转为实际值；
• 而pandas.read_spss()（pandas内置的SPSS读取方法）没有这个参数，对应功能需用convert_categoricals=True（旧版pandas）或结合分类类型参数处理。

修正方案二选一： #

方案A：使用pyreadstat库（推荐，参数匹配）

确保Databricks集群已安装pyreadstat，调用正确函数：

import pyreadstat

# 读取.sav文件，保留apply_value_labels参数
df, meta = pyreadstat.read_sav("/path/to/your/file.sav", apply_value_labels=True)
# 转为Spark DataFrame
spark_df = spark.createDataFrame(df)

方案B：使用pandas内置方法

若依赖pandas，移除apply_value_labels，改用pandas支持的参数：

import pandas as pd

# pandas读取SPSS文件，转换分类变量
df = pd.read_spss("/path/to/your/file.sav", convert_categoricals=True)
spark_df = spark.createDataFrame(df)

2. 解决iterrows()速度过慢问题 #

iterrows()是逐行迭代DataFrame，性能极差，尤其处理大文件时。优化方向是避免逐行迭代，改用向量化操作或高效批量处理：

优化方案： #

方案1：直接转换为Spark DataFrame（首选）

跳过逐行循环，直接将读取的DataFrame转为Spark DataFrame，利用分布式计算能力：

# 读取.sav文件后直接转Spark DF
df, meta = pyreadstat.read_sav("/path/to/your/file.sav", apply_value_labels=True)
spark_df = spark.createDataFrame(df)

# 后续用Spark API处理，比如保存到Delta表
spark_df.write.format("delta").mode("overwrite").saveAsTable("your_table_name")

方案2：改用itertuples()替代iterrows()

若必须行级处理，itertuples()返回元组而非Series，速度比iterrows()快5-10倍：

df, meta = pyreadstat.read_sav("/path/to/your/file.sav", apply_value_labels=True)
for row in df.itertuples(index=False):
    # 处理行数据，比如提取字段值
    col1_value = row.col1
    col2_value = row.col2
    # 批量收集数据后再提交到Spark，避免单条写入

方案3：Spark UDF批量处理（多文件场景）

若需处理大量.sav文件，将文件路径作为DataFrame列，用UDF批量读取：

from pyspark.sql.functions import udf
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
import pyreadstat

# 根据.sav文件结构定义Schema
custom_schema = StructType([
    StructField("col1", IntegerType(), True),
    StructField("col2", StringType(), True)
])

# 定义读取.sav文件的UDF
@udf(returnType=custom_schema)
def read_sav_file(file_path):
    df, _ = pyreadstat.read_sav(file_path, apply_value_labels=True)
    return df.iloc[0].to_dict()

# 创建包含文件路径的DataFrame
file_paths_df = spark.createDataFrame([("/dbfs/tmp/unzipped/file1.sav",), ("/dbfs/tmp/unzipped/file2.sav",)], ["file_path"])

# 批量读取处理
result_df = file_paths_df.withColumn("data", read_sav_file("file_path")).select("data.*")

额外注意事项 #

• Databricks依赖安装：若集群未安装pyreadstat，可在Notebook运行：

  %pip install pyreadstat
  

• 临时文件清理：处理完后清理解压的临时目录，避免占用存储空间：

  import shutil
  shutil.rmtree("/dbfs/tmp/your_temp_dir")
  

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