如何让Lambda导出Athena查询结果时忽略.csv.metadata文件

嗨，我来帮你搞定这个需求！你的Lambda函数现在能查询Athena并导出结果，但想去掉自动生成的.csv.metadata文件对吧？我给你完善代码并拆解关键步骤：

首先，咱们先把你的wait_for_result函数补全——原代码没处理查询失败/取消的情况，很容易陷入无限循环，而且要返回结果的S3路径方便后续操作：

import time
import boto3

def wait_for_result(athena, query_id):
    while True:
        query_execution = athena.get_query_execution(QueryExecutionId=query_id)
        state = query_execution['QueryExecution']['Status']['State']
        
        if state == 'SUCCEEDED':
            # 返回Athena结果的S3存储路径
            return query_execution['QueryExecution']['ResultConfiguration']['OutputLocation']
        elif state in ['FAILED', 'CANCELLED']:
            # 捕获查询失败/取消的错误信息
            error_msg = query_execution['QueryExecution']['Status'].get('StateChangeReason', 'Unknown error')
            raise Exception(f"Query ended with state {state}: {error_msg}")
        
        print(f'Query state: {state}, waiting 5s...')
        time.sleep(5)

接下来，添加处理S3文件的逻辑。这里分两种常见场景：

场景1：保留原路径的.csv文件，删除对应的.metadata文件 #

如果只是想清理Athena输出目录里的.metadata文件，用这个函数：

def clean_up_athena_metadata(s3, output_s3_path):
    # 解析S3路径：从"s3://bucket/path/result.csv"拆分桶名和前缀
    bucket_name, prefix = output_s3_path.replace('s3://', '').split('/', 1)
    
    # 列出该前缀下的所有文件（Athena会生成result.csv和result.csv.metadata）
    response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
    
    for obj in response.get('Contents', []):
        obj_key = obj['Key']
        # 只删除.metadata后缀的文件
        if obj_key.endswith('.csv.metadata'):
            print(f"Deleting metadata file: s3://{bucket_name}/{obj_key}")
            s3.delete_object(Bucket=bucket_name, Key=obj_key)

场景2：把.csv文件复制到指定目标存储桶（可选清理原文件） #

如果需要把.csv文件转移到另一个存储桶，同时忽略.metadata，用这个函数：

def copy_csv_to_target(s3, output_s3_path, target_bucket, target_folder=''):
    bucket_name, prefix = output_s3_path.replace('s3://', '').split('/', 1)
    response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
    
    for obj in response.get('Contents', []):
        obj_key = obj['Key']
        # 只处理.csv后缀的文件
        if obj_key.endswith('.csv'):
            # 构造目标路径：如果指定了目标文件夹，就把文件放到该文件夹下
            target_key = f"{target_folder}{obj_key.split('/')[-1]}" if target_folder else obj_key.split('/')[-1]
            
            print(f"Copying CSV file to target: s3://{target_bucket}/{target_key}")
            s3.copy_object(
                Bucket=target_bucket,
                Key=target_key,
                CopySource={'Bucket': bucket_name, 'Key': obj_key}
            )
            
            # 可选：复制完成后删除原路径的.csv和.metadata文件
            # s3.delete_object(Bucket=bucket_name, Key=obj_key)
            # s3.delete_object(Bucket=bucket_name, Key=f"{obj_key}.metadata")

整合到Lambda Handler #

最后把这些函数整合到Lambda的入口函数里，替换成你的实际参数：

def lambda_handler(event, context):
    # 初始化Athena和S3客户端
    athena = boto3.client('athena')
    s3 = boto3.client('s3')
    
    # 替换成你的查询执行ID（或者从event中动态获取）
    # 如果需要在Lambda里执行查询，可以添加athena.start_query_execution的代码
    query_id = "your-actual-query-execution-id"
    
    try:
        # 等待查询完成，获取结果路径
        output_path = wait_for_result(athena, query_id)
        print(f"Query succeeded! Output located at: {output_path}")
        
        # 选择你需要的处理方式（二选一或都用）
        # 方式1：清理原路径的metadata文件
        clean_up_athena_metadata(s3, output_path)
        
        # 方式2：复制CSV到目标存储桶
        # copy_csv_to_target(s3, output_path, "your-target-bucket-name", "target-folder/")
        
        return {
            'statusCode': 200,
            'body': 'Successfully processed Athena results - only CSV files retained.'
        }
    except Exception as e:
        print(f"Error occurred: {str(e)}")
        return {
            'statusCode': 500,
            'body': f'Failed to process results: {str(e)}'
        }

关键注意事项 #

• IAM权限：确保Lambda的IAM角色拥有以下权限：
  • athena:GetQueryExecution：用于查询执行状态
  • s3:ListBucket：用于列出Athena输出目录的文件
  • s3:DeleteObject（如果用场景1）：用于删除metadata文件
  • s3:CopyObject（如果用场景2）：用于复制CSV文件到目标桶

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