如何在PySpark中识别long类型列并将其转换为int类型？

原参考代码无法正常运行的核心原因有两点：

• cast() 方法需要传入目标数据类型的标准类型字符串/Spark类型对象，原代码传入的"to_timestamp"是函数名而非合法类型参数，本身逻辑存在错误
• 原代码的字段筛选逻辑是匹配时间类字段，和筛选long类型列的需求不匹配

可直接运行的实现代码 #

from pyspark.sql.functions import col

# 筛选出DataFrame中所有数据类型为long的列
long_columns = [col_name for col_name, col_type in df.dtypes if col_type == "long"]

# 遍历所有long类型列，统一转换为int类型
for col_name in long_columns:
    df = df.withColumn(col_name, col(col_name).cast("int"))

转换前注意事项 #

• Spark中Int类型为32位有符号整数，支持的取值范围为 -2147483648 ~ 2147483647，如果long列中存在超出该范围的数值，转换时会出现null值或数据溢出问题。
• 建议转换前先对所有long列做取值范围校验，避免数据异常，校验代码参考如下：

import pyspark.sql.functions as func

# 统计所有long列的最大值、最小值
check_result = df.select([
    func.max(col(c)).alias(f"{c}_max") for c in long_columns
] + [
    func.min(col(c)).alias(f"{c}_min") for c in long_columns
]).first()

# 逐列判断是否存在越界值
for col_name in long_columns:
    col_max = check_result[f"{col_name}_max"]
    col_min = check_result[f"{col_name}_min"]
    if col_max > 2147483647 or col_min < -2147483648:
        print(f"列 {col_name} 存在超出int取值范围的值，无法安全转换，列最大值：{col_max}，列最小值：{col_min}")

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