如何在PySpark DataFrame中自动转换各列数据类型

方案1：读取阶段直接开启自动类型推断（最简便） #

读取CSV文件时直接开启内置的类型推断参数，不需要读取后再逐列修改，是最省代码的方案：

df = spark.read.csv(
    "你的CSV文件路径",
    header=True,
    inferSchema=True, # 核心参数：开启自动扫描数据匹配列类型
    encoding="utf-8"
)
# 打印schema核对转换结果
df.printSchema()

注意：该方案会对数据集做一次扫描判断类型，小数据集无性能影响，TB级超大数据集不建议使用，会拖慢读取速度。

方案2：已读取为全String类型的DataFrame批量自动转换 #

如果你已经把数据读成了全String类型的DataFrame、不想重读，可以写通用批量转换逻辑，不用逐列手动写cast规则：

• 实现逻辑：对每列取少量非空样本值，按「整数->浮点数->布尔值->日期时间->字符串」的优先级尝试转换，匹配到对应类型就批量对整列做cast
• 可直接复用的代码如下：

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType, DoubleType, BooleanType, TimestampType

# 可根据自己数据集的实际日期格式补充
COMMON_DATE_FORMATS = [
    "yyyy-MM-dd", "yyyy/MM/dd", "dd/MM/yyyy", "MM-dd-yyyy",
    "yyyy-MM-dd HH:mm:ss", "yyyy/MM/dd HH:mm:ss"
]

def get_auto_cast_expr(spark_df, col_name):
    # 取100条非空样本做类型判断，避免扫描全表
    sample_data = [
        row[col_name] 
        for row in spark_df.select(col_name)
        .where(F.col(col_name).isNotNull())
        .limit(100)
        .collect()
    ]
    if not sample_data:
        return F.col(col_name)
    
    # 尝试匹配整数
    try:
        [int(val) for val in sample_data]
        return F.col(col_name).cast(IntegerType()).alias(col_name)
    except:
        pass
    
    # 尝试匹配浮点数
    try:
        [float(val) for val in sample_data]
        return F.col(col_name).cast(DoubleType()).alias(col_name)
    except:
        pass
    
    # 尝试匹配布尔值
    bool_candidate_set = {"true", "false", "0", "1", "yes", "no"}
    if all(val.lower() in bool_candidate_set for val in sample_data):
        return F.col(col_name).cast(BooleanType()).alias(col_name)
    
    # 尝试匹配日期时间
    for fmt in COMMON_DATE_FORMATS:
        try:
            test_val = spark_df.select(
                F.to_timestamp(F.col(col_name), fmt).alias("test_col")
            ).where(F.col(col_name).isNotNull()).first()["test_col"]
            if test_val is not None:
                return F.to_timestamp(F.col(col_name), fmt).alias(col_name)
        except:
            continue
    
    # 所有类型都匹配失败则保留原String类型
    return F.col(col_name)

# 替换成你自己的全String类型DataFrame变量名
# spark_df = 你的全string类型df变量
# 批量生成转换表达式并执行
cast_expressions = [get_auto_cast_expr(spark_df, col) for col in spark_df.columns]
final_df = spark_df.select(*cast_expressions)
# 打印结果核对
final_df.printSchema()

注意事项 #

• 自动转换依赖样本值判断，如果某列前100条样本都是整数、但后续存在非数值脏数据，转换后该部分脏数据会变成null，建议转换后做null值校验
• 如果列值包含千分位逗号、货币符号、全角数字这类特殊格式，要先做清洗再执行自动转换，否则会识别失败
• 对数据类型精度要求高的生产场景，还是建议提前定义明确的StructType schema，避免自动推断带来的类型不符合预期问题

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