如何在PySpark或AWS Glue Dynamic Frames中重命名生成的Parquet输出文件

Spark分布式计算的特性决定了默认输出会生成多个带随机前缀的part文件，要自定义固定文件名可以通过以下几种方案实现：

• 方案一：输出后重命名（最常用，不影响分布式性能）

  1. 先按常规方式把Parquet文件输出到S3的临时目录，比如s3://bucket/tmp-output/
  2. 用boto3调用S3的copy操作，把生成的目标数据文件复制到目标路径并重命名，可选择同步删除临时目录文件
     参考代码：

  import boto3
  from pyspark.context import SparkContext
  from awsglue.context import GlueContext
  
  glueContext = GlueContext(SparkContext.getOrCreate())
  s3 = boto3.client('s3')
  bucket_name = "替换为你的bucket名称"
  tmp_prefix = "tmp-output/"
  target_file_name = "abcd.parquet"
  
  # 1. 先将动态帧写出到临时目录
  dyf.write.parquet(f"s3://{bucket_name}/{tmp_prefix}", mode="overwrite")
  
  # 2. 过滤找到生成的Parquet数据文件，忽略_SUCCESS等标识文件
  response = s3.list_objects_v2(Bucket=bucket_name, Prefix=tmp_prefix)
  for obj in response['Contents']:
      if obj['Key'].endswith('.parquet'):
          # 复制重命名到目标路径
          s3.copy_object(
              Bucket=bucket_name,
              Key=target_file_name,
              CopySource={'Bucket': bucket_name, 'Key': obj['Key']}
          )
          # 可选：删除临时目录的原始文件
          s3.delete_object(Bucket=bucket_name, Key=obj['Key'])
          break
  

  适用场景：输出数据量不大、单文件即可承载的场景

• 方案二：单分区写出后改名（适合极小数据量）
  如果数据量很小，可以先把动态帧/数据框转为单分区，写出后再按方案一的步骤重命名，避免出现多个part文件
  参考代码：

  # 将数据重分区为1个分片
  single_partition_dyf = dyf.repartition(1)
  # 写出到临时目录后再按方案一的步骤重命名即可
  single_partition_dyf.write.parquet(f"s3://{bucket_name}/{tmp_prefix}", mode="overwrite")
  

  注意：仅建议数据量小于1GB时使用，重分区为1会把所有数据拉到同一个Executor处理，大数据量下容易出现内存溢出问题

• 方案三：使用Hadoop的FileUtil实现改名（不需要额外调用boto3）
  可以直接用PySpark内置的Hadoop API实现文件移动重命名，不需要额外引入boto3依赖
  参考代码：

  sc = SparkContext.getOrCreate()
  hadoop_conf = sc._jsc.hadoopConfiguration()
  fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)
  
  tmp_path = sc._jvm.org.apache.hadoop.fs.Path(f"s3://{bucket_name}/{tmp_prefix}")
  target_path = sc._jvm.org.apache.hadoop.fs.Path(f"s3://{bucket_name}/{target_file_name}")
  
  # 遍历临时目录找到parquet文件
  for file_status in fs.listStatus(tmp_path):
      if file_status.getPath().getName().endswith('.parquet'):
          fs.rename(file_status.getPath(), target_path)
          break
  # 可选：删除临时目录
  fs.delete(tmp_path, True)
  

• 大数据量下不建议强制生成单个自定义名称的Parquet文件，会失去分布式存储的优势，后续查询性能也会明显下降
• 如果是分区存储场景，建议仅对每个分区内的文件做自定义命名，不要合并全量数据为单文件

内容的提问来源于stack exchange，提问作者Gaurav Singh