移除S3存储Parquet文件表头空格的低算力实现方案

以下是几种算力消耗从低到高的解决方案，按需选择：

方案1：Athena表映射（算力消耗最低，无需修改原文件） #

这是最省资源的方式——通过Athena外部表的列名映射，直接实现无空格表头的查询，完全不触碰Parquet文件本身。

• 操作步骤：
  1. 先创建映射原带空格表头的外部表：

     CREATE EXTERNAL TABLE IF NOT EXISTS original_parquet (
         `I D` INT,
         `NA M E` STRING
         -- 按实际列定义补充其他字段
     )
     STORED AS PARQUET
     LOCATION 's3://your-bucket/target-path/';
     

  2. 再创建一个使用无空格列名的外部表，指向同一个S3路径：

     CREATE EXTERNAL TABLE IF NOT EXISTS cleaned_parquet (
         ID INT,
         NAME STRING
         -- 对应原字段的无空格名称
     )
     STORED AS PARQUET
     LOCATION 's3://your-bucket/target-path/';
     

  3. 后续直接查询cleaned_parquet即可，实际文件未被修改，仅通过元数据映射实现需求。

方案2：AWS Glue Serverless ETL（低算力，批量修改文件） #

利用Glue的serverless特性，仅处理列名映射，无需全量加载数据到内存，算力消耗可控。

• 操作步骤：
  1. 创建Glue爬虫，爬取目标S3路径的Parquet文件schema。
  2. 在Glue Studio中创建Python作业，使用以下代码批量重命名列：

     from awsglue.dynamicframe import DynamicFrame
     from pyspark.sql.functions import col
     
     # 读取S3中的Parquet文件
     raw_df = glueContext.create_dynamic_frame.from_options(
         connection_type="s3",
         connection_options={"paths": ["s3://your-bucket/source-path/"]},
         format="parquet"
     ).toDF()
     
     # 批量移除列名中的空格
     cleaned_df = raw_df.select(
         *[col(col_name).alias(col_name.replace(" ", "")) for col_name in raw_df.columns]
     )
     
     # 将修改后的数据写回S3（可覆盖原路径或写入新路径）
     glueContext.write_dynamic_frame.from_options(
         frame=DynamicFrame.fromDF(cleaned_df, glueContext, "cleaned_df"),
         connection_type="s3",
         connection_options={"path": "s3://your-bucket/dest-path/"},
         format="parquet"
     )
     

  3. 配置作业使用最小规格的DPU（如2个），降低算力消耗。

方案3：PyArrow轻量处理（本地/EC2低算力） #

用PyArrow高效操作Parquet元数据和流式处理数据，避免全量加载，适合小规模文件或自定义部署场景。

• 操作步骤：
  1. 安装依赖：pip install pyarrow boto3
  2. 运行以下代码处理S3中的Parquet文件：

     import pyarrow.parquet as pq
     import boto3
     from io import BytesIO
     
     s3_client = boto3.client('s3')
     BUCKET = "your-bucket"
     PREFIX = "path/to/parquet-files/"
     
     # 遍历S3中的Parquet文件
     paginator = s3_client.get_paginator('list_objects_v2')
     for page in paginator.paginate(Bucket=BUCKET, Prefix=PREFIX):
         for obj in page.get('Contents', []):
             file_key = obj['Key']
             if not file_key.endswith('.parquet'):
                 continue
     
             # 下载文件到内存缓冲区
             response = s3_client.get_object(Bucket=BUCKET, Key=file_key)
             input_buffer = BytesIO(response['Body'].read())
     
             # 读取原schema并修改列名
             parquet_file = pq.ParquetFile(input_buffer)
             old_schema = parquet_file.schema
             new_col_names = [name.replace(" ", "") for name in old_schema.names]
             new_schema = old_schema.rename_columns(new_col_names)
     
             # 流式写入修改后的Parquet文件
             output_buffer = BytesIO()
             with pq.ParquetWriter(output_buffer, new_schema) as writer:
                 for batch in parquet_file.iter_batches():
                     writer.write_batch(batch.rename_columns(new_col_names))
     
             # 写回S3（可覆盖原文件或写入新路径）
             s3_client.put_object(Bucket=BUCKET, Key=file_key, Body=output_buffer.getvalue())
     

  3. 选择t2.micro这类轻量EC2实例或本地机器运行，控制算力成本。

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