如何让PySpark DataFrame正确识别CSV与Parquet文件的列数据类型？

一、针对CSV文件的优化读取方式 #

CSV属于无内置Schema的文本格式，PySpark默认会将所有列识别为字符串类型，以下是两种自动推断/转换类型的实用方案：

1. 启用内置类型推断
   直接通过inferSchema=True参数让PySpark自动扫描数据样本推断类型：

df = spark.read.option("header", "true") \
               .option("inferSchema", "true") \
               .csv("path/to/csv/files")

注意：该方式会额外扫描一次数据，超大文件可能影响性能；若列内存在混合类型（如同一列既有整数又有文本），推断结果可能不准确。

2. 通用自动类型转换函数
   如果内置推断效果不佳，可编写通用函数，安全地将字符串列尝试转换为更合适的类型（优先级：整数→浮点数→日期）：

from pyspark.sql.functions import col, try_cast
from pyspark.sql.types import StringType, IntegerType, DoubleType, DateType

def auto_cast_columns(df):
    for column in df.columns:
        current_type = df.schema[column].dataType
        if isinstance(current_type, StringType):
            # 尝试转换为整数
            int_df = df.withColumn(column, try_cast(col(column), IntegerType()))
            # 仅当转换后空值数量未增加（所有非空值都能转整数）时生效
            if int_df.select(column).filter(col(column).isNull()).count() == df.select(column).filter(col(column).isNull()).count():
                df = int_df
                continue
            # 尝试转换为浮点数
            double_df = df.withColumn(column, try_cast(col(column), DoubleType()))
            if double_df.select(column).filter(col(column).isNull()).count() == df.select(column).filter(col(column).isNull()).count():
                df = double_df
                continue
            # 尝试转换为日期（可根据需求自定义日期格式）
            date_df = df.withColumn(column, try_cast(col(column), DateType()))
            if date_df.select(column).filter(col(column).isNull()).count() == df.select(column).filter(col(column).isNull()).count():
                df = date_df
    return df

# 使用示例：先读取CSV（默认字符串类型），再执行自动转换
raw_csv_df = spark.read.option("header", "true").csv("path/to/csv/files")
casted_csv_df = auto_cast_columns(raw_csv_df)

该函数通过try_cast实现安全转换，仅在所有非空值都能成功转换时才替换列类型，避免数据丢失。

二、针对Parquet文件的处理 #

Parquet是带内置Schema的列式存储格式，正常读取时会直接沿用文件定义的类型。若出现类型识别错误，通常是因为写入Parquet时就使用了错误类型，可通过以下方式处理：

1. 读取后应用自动转换函数
   直接使用上述auto_cast_columns函数对读取后的Parquet DataFrame进行类型修正：

raw_parquet_df = spark.read.parquet("path/to/parquet/files")
corrected_parquet_df = auto_cast_columns(raw_parquet_df)

2. 强制指定Schema读取
   若已知正确的Schema，可提前定义并强制读取：

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

custom_schema = StructType([
    StructField("user_id", IntegerType(), nullable=True),
    StructField("user_name", StringType(), nullable=True)
])

df = spark.read.schema(custom_schema).parquet("path/to/parquet/files")

三、通用注意事项 #

• 若列内存在混合类型（如同一列既有数字又有文本），建议保留字符串类型，避免转换导致数据丢失。
• 自动推断/转换后，务必通过df.printSchema()和df.show()验证结果准确性。
• 超大规模数据集优先使用预定义Schema，兼顾性能与准确性。

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