如何使用Python删除关联S3多文件的AWS Glue表特定记录？

AWS Glue本身不支持直接行级删除记录——它是基于S3对象存储的无服务器数据目录，数据实际存储在S3的文件中。要删除特定记录，只能通过读取文件→过滤目标记录→重写干净文件的方式实现，以下是通用的Python实现方案，支持CSV、JSON等常见格式。

核心思路 #

1. 从Glue元数据中获取表的S3存储路径、数据格式、分区规则（如果有）
2. 遍历S3中所有数据文件，逐个处理：
   • 下载文件到本地临时空间
   • 读取并过滤掉需要删除的记录
   • 若过滤后仍有数据，将干净数据写回原S3路径；若文件为空则直接删除原文件
3. （可选）更新Glue表的统计信息，确保查询准确性

具体实现代码 #

1. 依赖安装 #

先安装必要的库：

pip install boto3 pandas s3fs tempfile

2. 完整代码 #

import boto3
import pandas as pd
import tempfile
import os
from s3fs import S3FileSystem
from typing import Callable

# 初始化客户端（替换为你的AWS区域）
glue_client = boto3.client('glue', region_name='us-east-1')
s3 = S3FileSystem()

def get_glue_table_details(database_name: str, table_name: str) -> dict:
    """获取Glue表的核心元数据：S3路径、数据格式、分区列"""
    response = glue_client.get_table(DatabaseName=database_name, Name=table_name)
    table = response['Table']
    return {
        's3_path': table['StorageDescriptor']['Location'],
        'input_format': table['StorageDescriptor']['InputFormat'],
        'columns': [col['Name'] for col in table['StorageDescriptor']['Columns']],
        'partition_keys': [pk['Name'] for pk in table.get('PartitionKeys', [])]
    }

def filter_csv_file(temp_file_path: str, filter_func: Callable) -> pd.DataFrame:
    """过滤CSV文件中的目标记录"""
    df = pd.read_csv(temp_file_path)
    return df[~filter_func(df)]

def filter_json_file(temp_file_path: str, filter_func: Callable) -> pd.DataFrame:
    """过滤JSON文件中的目标记录（支持单行/多行JSON）"""
    df = pd.read_json(temp_file_path, lines=True)
    return df[~filter_func(df)]

def process_s3_file(s3_file_path: str, filter_func: Callable, data_format: str):
    """处理单个S3文件：下载→过滤→重写/删除"""
    # 创建临时文件
    with tempfile.NamedTemporaryFile(mode='r+', delete=False) as temp_file:
        temp_path = temp_file.name
    
    try:
        # 下载S3文件到临时路径
        s3.download(s3_file_path, temp_path)
        
        # 根据格式过滤数据
        filtered_df = None
        if data_format in ['org.apache.hadoop.mapred.TextInputFormat']:
            # 适配CSV格式
            filtered_df = filter_csv_file(temp_path, filter_func)
        elif data_format in ['org.apache.hadoop.mapreduce.lib.input.TextInputFormat']:
            # 适配单行JSON格式
            filtered_df = filter_json_file(temp_path, filter_func)
        else:
            raise ValueError(f"不支持的数据格式: {data_format}")
        
        # 处理过滤后的数据
        if filtered_df.empty:
            # 无剩余数据，删除原S3文件
            s3.rm(s3_file_path)
            print(f"文件已删除: {s3_file_path}")
        else:
            # 将干净数据写回原路径（覆盖原文件）
            if data_format == 'org.apache.hadoop.mapred.TextInputFormat':
                filtered_df.to_csv(temp_path, index=False)
            else:
                filtered_df.to_json(temp_path, orient='records', lines=True)
            s3.upload(temp_path, s3_file_path)
            print(f"文件已更新: {s3_file_path}")
    finally:
        # 清理临时文件
        os.unlink(temp_path)

def delete_glue_table_records(database_name: str, table_name: str, filter_func: Callable):
    """主函数：删除Glue表中符合条件的记录"""
    # 获取表详情
    table_details = get_glue_table_details(database_name, table_name)
    s3_path = table_details['s3_path']
    data_format = table_details['input_format']
    
    # 遍历S3路径下的所有数据文件（跳过目录和元数据文件）
    for s3_file in s3.glob(f"{s3_path}/**/*"):
        if s3.isdir(s3_file) or s3_file.endswith('/') or s3_file.endswith('.metadata'):
            continue
        process_s3_file(s3_file, filter_func, data_format)

# ------------------- 使用示例 -------------------
if __name__ == '__main__':
    # 自定义过滤规则：删除user_id=123的记录
    def filter_condition(df):
        return df['user_id'] == 123
    
    # 执行删除操作（替换为你的库名和表名）
    delete_glue_table_records(
        database_name='your-database-name',
        table_name='your-table-name',
        filter_func=filter_condition
    )

关键说明 #

• 过滤规则自定义：filter_condition函数根据业务需求编写，返回布尔值（True代表需要删除的记录）
• 格式扩展：若需要支持Parquet等格式，可添加对应的过滤函数，用pd.read_parquet/df.to_parquet处理
• 分区表优化：如果是分区表，可在遍历S3文件时指定目标分区路径（比如s3.glob(f"{s3_path}/year=2024/month=05/**/*")），减少不必要的文件处理
• 性能与风险：单进程Python适合中小规模数据，大规模数据建议用AWS Glue Job（Spark）实现；重写文件前可先备份原文件到S3备份目录，避免数据丢失

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