PySpark对表中日期字段排序时出现Parquet相关报错如何解决

该错误属于Parquet文件存储类型与预期读取类型不匹配引发的解析异常，核心触发原因有两个：

• Glue Catalog中定义的process_date字段类型，和底层Parquet文件实际存储的类型不一致：比如元数据中标注该字段为字符串/日期类型，但实际Parquet中该字段以Integer格式存储（例如时间戳、yyyyMMdd格式的数值），排序操作触发类型转换时，Parquet的字典编码解析失败抛出异常。
• 过滤逻辑中手动拼接单引号强制将匹配值转为字符串，若process_date实际为整型，类型不匹配会进一步加重解析错误。

前置校验 #

首先执行以下代码确认process_date的实际Schema：

raw_customers.printSchema()

根据返回的字段类型选择对应修复方案：

方案1：修正字段类型与过滤逻辑 #

如果确认process_date实际为Integer类型，去掉过滤逻辑中的单引号，或直接使用列对象匹配避免类型问题：

from pyspark.sql.functions import desc

raw_customers = glueContext.create_dynamic_frame.from_catalog(database = "postgresql_processed", table_name = "prodplfm_plf_customers").toDF()
# 可选：根据业务需要统一转换process_date为指定类型，比如转日期型
# raw_customers = raw_customers.withColumn("process_date", raw_customers["process_date"].cast("date"))
latest_partition=raw_customers.select("process_date").orderBy(desc("process_date")).limit(1).collect()[0][0]
# 直接使用列匹配，避免手动拼接字符串导致的类型问题
customers=raw_customers.filter(raw_customers.process_date == latest_partition)
customers.createOrReplaceTempView("customers")

方案2：修复元数据与Schema不一致问题 #

如果是Glue元数据和实际Parquet类型不匹配，可选择以下两种方式修复：

• 登录Glue控制台，找到对应表修正process_date的字段类型和实际存储类型一致
• 读取时开启Schema合并自动适配实际存储类型：

raw_customers = glueContext.create_dynamic_frame.from_catalog(
    database = "postgresql_processed", 
    table_name = "prodplfm_plf_customers",
    additional_options = {"mergeSchema": "true"}
).toDF()

方案3：优化实现避免全表扫描（推荐） #

如果仅需要读取最新分区数据，不需要全表扫描排序，直接调用Glue接口获取分区元数据即可，性能更高且不会触发字段解析错误：

import boto3
from pyspark.sql.functions import desc

glue_client = boto3.client('glue')
# 直接读取表的分区元数据
partition_resp = glue_client.get_partitions(
    DatabaseName="postgresql_processed",
    TableName="prodplfm_plf_customers"
)
# 排序取最新分区值
all_partitions = [p["Values"][0] for p in partition_resp["Partitions"]]
latest_partition = sorted(all_partitions, reverse=True)[0]
# 下推算子直接过滤分区，不需要加载全表数据
customers = glueContext.create_dynamic_frame.from_catalog(
    database = "postgresql_processed", 
    table_name = "prodplfm_plf_customers",
    push_down_predicate = f"process_date='{latest_partition}'"
).toDF()
customers.createOrReplaceTempView("customers")

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