在Azure Databricks中用PySpark实现Parquet字段通用自动类型转换

针对从PostgreSQL导出、所有列均为VARCHAR类型的Parquet文件，可通过以下通用方法在Azure Databricks中自动识别并转换为实际数据类型：

核心思路 #

遍历DataFrame的每一列，按优先级尝试转换为常见数据类型（时间戳/日期 > 整数 > 小数 > 字符串），转换失败则保留原字符串值，无需针对不同Schema编写定制化代码。

实现步骤 #

1. 读取ADLS上的Parquet文件 #

先将文件加载为Spark DataFrame，此时所有列默认是string类型：

df = spark.read.parquet("abfss://<container-name>@<storage-account>.dfs.core.windows.net/<parquet-file-path>")

2. 定义自动转换函数 #

利用Spark内置的try_cast函数（转换失败返回null），结合when/otherwise实现类型优先级判断：

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

def auto_cast_column(col_name):
    # 尝试转换为时间戳（优先级最高，覆盖日期格式的字符串）
    casted_ts = F.try_cast(F.col(col_name), TimestampType())
    # 尝试转换为日期
    casted_date = F.try_cast(F.col(col_name), DateType())
    # 尝试转换为整数
    casted_int = F.try_cast(F.col(col_name), IntegerType())
    # 尝试转换为小数
    casted_double = F.try_cast(F.col(col_name), DoubleType())
    
    # 按优先级返回转换结果，失败则保留原字符串
    return F.when(casted_ts.isNotNull(), casted_ts) \
            .when(casted_date.isNotNull(), casted_date) \
            .when(casted_int.isNotNull(), casted_int) \
            .when(casted_double.isNotNull(), casted_double) \
            .otherwise(F.col(col_name)).alias(col_name)

3. 批量应用转换到所有列 #

遍历DataFrame的所有列，调用转换函数生成新的DataFrame：

auto_casted_df = df.select([auto_cast_column(col) for col in df.columns])

4. 验证转换结果 #

打印Schema和样例数据，确认类型转换是否符合预期：

# 查看转换后的Schema
auto_casted_df.printSchema()
# 查看前5条数据（不截断内容）
auto_casted_df.show(5, truncate=False)

自定义优化建议 #

• 调整类型优先级：如果业务中数值类型优先级高于日期，可调整when的顺序
• 指定日期格式：针对非标准日期格式，替换try_cast为F.to_date(F.col(col_name), "yyyy-MM-dd HH:mm:ss")，并增加格式校验
• 处理高精度小数：若需保留小数精度，将DoubleType替换为DecimalType(precision=18, scale=6)
• 排除特定列：如果部分列无需转换，可过滤列列表：

  exclude_cols = ["固定字符串列1", "固定字符串列2"]
  target_cols = [col for col in df.columns if col not in exclude_cols]
  auto_casted_df = df.select(
      [auto_cast_column(col) for col in target_cols] + 
      [F.col(col) for col in exclude_cols]
  )
  

注意事项 #

• try_cast仅在Spark 2.3及以上版本支持，低版本可通过UDF实现，但性能略差
• 转换过程不会修改原始文件，需将转换后的DataFrame重新写入ADLS：auto_casted_df.write.parquet("<目标路径>", mode="overwrite")
• 对于模糊的字符串（如"20230101"既可以解析为日期也可以解析为整数），需根据业务逻辑确定优先级

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