单文件处理分区Parquet表行级转换：是否存在风险及元数据损失？

我在S3上存储了一张大型分区Parquet表，结构如下：

s3
├── gender=M
│	├── age=10
│	│	├── part-000.parquet
│	│	├── part-001.parquet
│	│	└── part-002.parquet
│	└── age=20
│		├── part-000.parquet
│		├── part-001.parquet
│		└── part-002.parquet
└── gender=W
	├── age=10
	│	├── part-000.parquet
	│	├── part-001.parquet
	│	└── part-002.parquet
	└── age=20
		├── part-000.parquet
		├── part-001.parquet
		└── part-002.parquet

我需要对部分列执行行级转换，转换不会改变表的Schema/维度，也不依赖当前行以外的数据（例如给字符串列添加指定前缀）。由于分区数据量极大，我采用逐个文件处理的方案：将原文件移至临时位置，用Spark读取转换，再将转换后的文件移回原位置覆盖原文件，伪代码流程如下：

以文件s3://gender=M/age=10/part-000.parquet为例：

1. 将文件移至临时位置：

's3://gender=M/age=10/part-000.parquet' ⟼ 's3://tmp/gender=M/age=10/part-000.parquet'

2. 用Spark读取临时文件并执行转换：

df = spark.read.parquet('s3://tmp/gender=M/age=10/part-000.parquet')
transformed_df = transform(df)

3. 将转换后的DataFrame写入新临时位置：

transformed_df.coalesce(1).write.parquet('s3://tmp/transformed/gender=M/age=10/')
# 生成的文件名可能为s3://tmp/transformed/gender=M/age=10/part-xyz.parquet

4. 将转换后的文件移回原位置并使用原文件名：

's3://tmp/transformed/gender=M/age=10/part-xyz.parquet' ⟼ 's3://gender=M/age=10/part-000.parquet'

5. 删除临时目录s3://tmp/

测试后数据结果正确，但我不确定该方案是否会对后续数据使用造成问题，以及是否会丢失表的元数据。

补充说明：所有转换无需过滤数据，且均通过mapInArrow实现，该方法特性类似标准udfs，未利用Spark executor的并行能力。

一、元数据丢失风险 #

Parquet表的分区元数据（gender、age这类分区列信息）是通过目录结构维护的，你的方案全程保留了原有的分区目录结构，且转换后文件写回原位置时没有修改分区目录，因此分区元数据不会丢失。

至于Parquet文件内部的元数据：

• 你用Spark读取原文件后执行行级转换，转换不修改Schema，mapInArrow也保证了数据结构一致性，Spark写入时会自动生成符合规范的Parquet元数据（包括列类型、压缩信息等），和原文件的元数据逻辑一致，不会出现丢失或损坏的情况。
• 唯一需要注意：如果原文件有自定义的Parquet元数据（比如用户添加的键值对），Spark默认不会保留这类自定义元数据，需要在写入时显式配置spark.sql.parquet.writeLegacyFormat或通过额外代码处理，但如果你的场景没有自定义元数据，就无需担心。

二、后续数据使用的潜在问题 #

1. 原子性风险
   你的方案中，从原文件移走、转换到写回的过程不是原子操作：

   • 如果在步骤1完成后、步骤4写回前，有其他程序读取该分区，会出现文件缺失的情况，导致读取失败或数据不完整。
   • 若步骤4写回时失败，原文件在临时目录，原位置为空，会造成数据丢失。
     建议优化：先完成转换并把文件写到临时位置，再原子替换原文件（S3的mv操作在同一存储桶下是原子的），也就是调换步骤1和步骤3的顺序：先转换原文件到临时输出，再用mv原子替换原文件，避免中间状态的缺失。

2. 性能效率问题
   你用coalesce(1)强制生成单个文件，虽然保证了和原文件数量一致，但完全放弃了Spark的并行处理能力，加上mapInArrow没用到executor并行，单文件处理的效率极低，对于超大规模分区来说，这个方案的耗时会非常长。
   优化建议：

   • 取消coalesce(1)，允许Spark生成多个文件，之后再合并成单个文件（可以用S3的合并工具或Spark的repartition(1)，但repartition会 shuffle 数据，需要权衡）；
   • 尝试批量处理同一分区下的多个文件，而不是逐个处理，利用Spark的并行读取能力提升效率。

3. 数据一致性验证
   虽然测试结果正确，但大规模处理时建议添加校验步骤：

   • 转换后对比原文件和新文件的行数（因为无过滤，行数必须一致）；
   • 抽样验证转换后的数据内容是否符合预期；
   • 保留临时文件到所有分区处理完成后再删除，避免出错后无法恢复。

三、其他注意事项 #

• S3的最终一致性：写回文件后，可能需要短暂等待才能被其他读取程序看到，若有实时读取场景，需要考虑这一点。
• 权限问题：确保执行操作的角色有S3的读写、删除、移动权限，避免中途因权限失败导致数据异常。

内容的提问来源于stack exchange，提问作者L.B.