使用Boto3复制AWS S3文件时如何实现动态文件名命名？

核心原因 #

你在调用start_query_execution后立刻执行复制操作，此时Athena查询还未完成，S3上尚未生成对应的.csv结果文件，因此触发NoSuchKey错误。硬编码文件名能成功，是因为对应的查询已执行完毕，文件已存在。

解决步骤 #

1. 等待查询执行完成
   在获取Query ID后，需要循环检查查询状态，直到状态变为SUCCEEDED，再执行复制操作。

2. 修正代码潜在问题

   • 补充缺失的session初始化代码；
   • 确保RESULT_OUTPUT_LOCATION与拼接的src_key路径完全匹配，避免路径不一致导致文件找不到。

修改后完整代码 #

import boto3
import time

# 配置AWS参数
AWS_ACCESS_KEY = "你的Access Key"
AWS_SECRET_KEY = "你的Secret Key"
AWS_REGION = "你的AWS区域"
RESULT_OUTPUT_LOCATION = "s3://smg-datalake-prod-athena-query-results/python-athena/"

# 初始化AWS会话与客户端
session = boto3.Session(
    aws_access_key_id=AWS_ACCESS_KEY,
    aws_secret_access_key=AWS_SECRET_KEY,
    region_name=AWS_REGION
)
s3 = session.client('s3')
athena_client = session.client("athena")

def wait_for_query_completion(query_execution_id):
    """等待Athena查询完成，返回最终状态"""
    while True:
        response = athena_client.get_query_execution(QueryExecutionId=query_execution_id)
        status = response['QueryExecution']['Status']['State']
        if status in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
            return status
        time.sleep(2)  # 每2秒检查一次状态

def main():
    query = "select * from test_table"
    # 启动Athena查询
    response = athena_client.start_query_execution(
        QueryString=query,
        ResultConfiguration={"OutputLocation": RESULT_OUTPUT_LOCATION}
    )
    query_id = response['QueryExecutionId']
    
    # 等待查询完成，失败则终止流程
    status = wait_for_query_completion(query_id)
    if status != 'SUCCEEDED':
        print(f"查询执行失败，状态: {status}")
        return
    
    # 配置存储桶与文件名
    src_bucket = 'smg-datalake-prod-athena-query-results'
    dst_bucket = 'smg-datalake-prod-athena-query-results'
    file_name = f"{query_id}.csv"
    copy_object(src_bucket, dst_bucket, file_name)

def copy_object(src_bucket, dst_bucket, file_name):
    src_key = f'python-athena/{file_name}'
    dst_key = f'python-athena/cosmo/rss/v2/newsletter/kloka_latest.csv'
    try:
        s3.copy_object(
            Bucket=dst_bucket,
            CopySource={'Bucket': src_bucket, 'Key': src_key},
            Key=dst_key
        )
        print("文件复制成功")
    except Exception as e:
        print(f"复制失败: {str(e)}")

if __name__ == "__main__":
    main()

额外注意事项 #

• 若查询结果数据量较大，Athena会生成多个分片文件（如query_id.csv、query_id.csv.00001），如需复制所有结果，需遍历S3对应前缀下的所有文件；
• 确认RESULT_OUTPUT_LOCATION末尾包含斜杠，保证路径拼接后与Athena生成的文件路径完全一致；
• 代码中添加了异常处理，避免查询失败后执行无效的复制操作。

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