AWS Glue读取Parquet文件时缺失列问题求助

一、解决读取不到sex列的问题 #

核心原因是Parquet文件schema不一致，Glue默认仅读取首个文件的schema，导致含sex列的文件无法被正确解析，以下是修复步骤：

1. 确认数据路径与分区配置

   • 检查Glue表关联的S3路径，确保所有包含sex列的Parquet文件都在该路径下，若为分区表，需确认分区目录未遗漏目标文件。
   • 若用Glue爬虫建表，重新运行爬虫时勾选「Update all new and existing partitions with metadata from the table」，强制合并所有文件的schema。

2. 开启Schema合并读取配置
   无论是Glue Dynamic Frame还是Spark DataFrame读取，都需开启mergeSchema参数来合并不同文件的schema：

   • 使用Glue Dynamic Frame读取：

     from awsglue.context import GlueContext
     from pyspark.context import SparkContext
     
     sc = SparkContext()
     glueContext = GlueContext(sc)
     
     dynamic_frame = glueContext.create_dynamic_frame.from_catalog(
         database="你的数据库名",
         table_name="你的表名",
         additional_options={"mergeSchema": "true"}
     )
     

   • 使用Spark DataFrame读取：

     df = spark.read.parquet("s3://你的存储桶路径/", mergeSchema=True)
     

3. 验证并修正表Schema
   手动检查Glue表的Schema定义，确保sex列的类型（如string）与实际Parquet文件中的列类型一致。若不一致，直接在Glue控制台编辑表的Schema更新sex列定义。

二、实现数据标准化：合并gender与sex为gender_sex #

读取完整列数据后，按以下步骤统一输出格式：

1. 转换为DataFrame处理
   将Glue Dynamic Frame转换为Spark DataFrame，方便使用Spark函数做数据清洗：

   from pyspark.sql.functions import coalesce, col
   
   df = dynamic_frame.toDF()
   

2. 合并列并保留目标字段
   使用coalesce函数优先取非空值，合并gender和sex为gender_sex，再保留所需列：

   # 合并gender和sex，优先取非空值
   df = df.withColumn("gender_sex", coalesce(col("gender"), col("sex")))
   
   # 保留最终需要的三列
   df = df.select("id", "name", "gender_sex")
   

3. 转换回Dynamic Frame（可选）
   若需用Glue原生组件写入数据，转换回Dynamic Frame：

   from awsglue.dynamicframe import DynamicFrame
   
   final_dynamic_frame = DynamicFrame.fromDF(df, glueContext, "final_dynamic_frame")
   

三、大型ETL管道前置建议 #

针对后续扩展的大型管道，提前做好以下准备：

• 强制指定Schema：提前定义统一Schema并传入读取函数，避免Schema漂移，示例：

  from pyspark.sql.types import StructType, StructField, StringType
  
  custom_schema = StructType([
      StructField("id", StringType(), nullable=True),
      StructField("name", StringType(), nullable=True),
      StructField("gender", StringType(), nullable=True),
      StructField("sex", StringType(), nullable=True)
  ])
  
  df = spark.read.schema(custom_schema).parquet("s3://你的存储桶路径/", mergeSchema=True)
  

• 规划数据分区：按业务维度（如日期）设置分区，提升后续读取和处理效率。

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