如何处理Amazon Redshift Spectrum Parquet追加时_metadata文件膨胀与损坏问题

问题背景 #

Web应用生成大量含JSON字符串的.txt日志文件存储在S3，按每批200个文件执行以下处理流程：

• 用Python Pandas读取S3中的200个文件
• 提取数据保存至Amazon Redshift表
• 通过dataframe.to_parquet以追加模式按log_date分区保存为Parquet格式
• 借助Redshift Spectrum关联S3查询Parquet数据

执行第三步时，fastparquet生成的_metadata文件持续膨胀，超过70MB后损坏，导致无法继续追加数据。

可行解决方案 #

1. 跳过_metadata文件生成（保留追加模式） #

fastparquet的to_parquet方法支持通过参数禁用_metadata和_common_metadata的生成，直接写入分区文件。Redshift Spectrum无需这些元数据文件即可正常查询Parquet数据（会直接扫描分区内的Parquet文件），完全规避元数据膨胀问题。

修改后的代码：

dataframe.to_parquet(
    path=s3_spectrum_loc,
    engine='fastparquet',
    index=False,
    partition_cols=['log_date'],
    times='int96',
    append=True,
    write_metadata=False,  # 禁用_metadata文件生成
    write_common_metadata=False  # 禁用_common_metadata文件生成
)

2. 手动控制元数据生成，避免自动累积 #

若需保留元数据文件但不想让其无限膨胀，可放弃append=True的自动元数据累积，改为手动写入文件+定期重建元数据：

• 将当前批次的Parquet文件写入临时路径
• 移动临时文件到正式分区目录
• 定期（如每日）重新生成全量元数据，覆盖旧文件

示例代码：

import boto3
from fastparquet import write, ParquetFile

s3_client = boto3.client('s3')

# 1. 写入当前批次到临时分区路径（S3示例）
temp_bucket = 'your-bucket'
temp_prefix = f"{s3_spectrum_loc}_temp/log_date={dataframe['log_date'].iloc[0]}"
dataframe.to_parquet(
    path=f"s3://{temp_bucket}/{temp_prefix}",
    engine='fastparquet',
    index=False,
    times='int96',
    write_metadata=False
)

# 2. 移动临时文件到正式路径
response = s3_client.list_objects_v2(Bucket=temp_bucket, Prefix=temp_prefix)
for obj in response.get('Contents', []):
    src_key = obj['Key']
    dst_key = src_key.replace('_temp/', '')
    s3_client.copy_object(
        Bucket=temp_bucket,
        CopySource={'Bucket': temp_bucket, 'Key': src_key},
        Key=dst_key
    )
    s3_client.delete_object(Bucket=temp_bucket, Key=src_key)

# 3. 每日执行一次，重新生成全量元数据
def regenerate_full_metadata(bucket, base_prefix):
    pf = ParquetFile(f"s3://{bucket}/{base_prefix}")
    pf.to_metadata_file(f"s3://{bucket}/{base_prefix}/_metadata")
    pf.to_common_metadata_file(f"s3://{bucket}/{base_prefix}/_common_metadata")

# 按需调用：regenerate_full_metadata('your-bucket', s3_spectrum_loc)

3. 切换到pyarrow引擎处理追加写入 #

pyarrow对Parquet元数据的处理逻辑与fastparquet不同，追加写入时不会无限累积元数据，仅更新必要信息，能有效控制文件大小。

修改后的代码：

dataframe.to_parquet(
    path=s3_spectrum_loc,
    engine='pyarrow',
    index=False,
    partition_cols=['log_date'],
    engine_kwargs={'use_deprecated_int96_timestamps': True},  # 对应fastparquet的times='int96'
    append=True
)

注：pyarrow的分区目录结构与fastparquet一致，Redshift Spectrum可无缝兼容。需提前安装pyarrow：pip install pyarrow

4. 按分区单独生成元数据 #

如果必须保留元数据，可按log_date分区单独生成元数据文件，单个分区的元数据大小可控，且不会互相影响。

示例代码：

# 按log_date分组，每个分区单独写入并生成元数据
for log_date, group_df in dataframe.groupby('log_date'):
    partition_path = f"{s3_spectrum_loc}/log_date={log_date}"
    group_df.to_parquet(
        path=partition_path,
        engine='fastparquet',
        index=False,
        times='int96',
        write_metadata=True,  # 只为当前分区生成元数据
        write_common_metadata=False
    )

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