AWS Lambda调用Athena查询时自定义S3输出CSV文件名的方案咨询

AWS Athena 原生不支持提交查询时直接指定输出CSV的文件名，默认生成的结果文件名为{QueryExecutionId}.csv，附带的元数据文件名为{QueryExecutionId}.csv.metadata，统一存放在你指定的OutputLocation路径下。可通过以下几种方案实现自定义文件名需求：

方案1：查询完成后重命名S3对象（最通用，适配所有场景） #

确认查询执行成功后，调用S3接口将默认生成的文件复制为你指定的文件名，按需删除原始文件即可。该方案无需修改查询逻辑，对结果集大小无限制。

优化后的完整代码示例： #

import boto3
import time

def lambda_handler(event, context):
    athena_client = boto3.client('athena')
    s3_client = boto3.client('s3')
    
    # 自定义配置项
    ATHENA_DATABASE = 'raw'
    S3_OUTPUT_BUCKET = 'my-bucket-name'
    CUSTOM_OUTPUT_FILE = 'campaign_ids.csv' # 你想要的输出文件名
    QUERY_SQL = "SELECT id from campaign;"

    # 提交Athena查询
    query_resp = athena_client.start_query_execution(
        QueryString = QUERY_SQL,
        QueryExecutionContext = {'Database': ATHENA_DATABASE},
        ResultConfiguration = {'OutputLocation': f's3://{S3_OUTPUT_BUCKET}'}
    )
    query_execution_id = query_resp['QueryExecutionId']
    
    # 轮询查询执行状态，替换原固定sleep逻辑，避免查询未完成导致报错
    while True:
        exec_state = athena_client.get_query_execution(QueryExecutionId=query_execution_id)['QueryExecution']['Status']['State']
        if exec_state in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
            break
        time.sleep(1)
    
    if exec_state != 'SUCCEEDED':
        raise RuntimeError(f"Athena查询执行失败，当前状态：{exec_state}")
    
    # 复制默认文件为自定义文件名
    source_file_key = f"{query_execution_id}.csv"
    s3_client.copy_object(
        Bucket = S3_OUTPUT_BUCKET,
        Key = CUSTOM_OUTPUT_FILE,
        CopySource = {'Bucket': S3_OUTPUT_BUCKET, 'Key': source_file_key}
    )
    
    # 可选操作：删除Athena生成的原始结果文件和元数据文件
    s3_client.delete_object(Bucket=S3_OUTPUT_BUCKET, Key=source_file_key)
    s3_client.delete_object(Bucket=S3_OUTPUT_BUCKET, Key=f"{source_file_key}.metadata")
    
    # 原业务逻辑
    results = athena_client.get_query_results(QueryExecutionId = query_execution_id)
    for row in results['ResultSet']['Rows']:
        print(row)

权限说明： #

需要给Lambda执行角色新增以下S3权限：

• s3:GetObject
• s3:PutObject
• s3:DeleteObject

方案2：自行拼装结果写入S3（适合小结果集场景） #

如果查询结果数据量较小（<100MB），可以直接遍历get_query_results返回的结果集，手动拼装为CSV格式后直接上传到S3指定路径，完全自主控制文件名。该方案无需处理Athena生成的默认文件，缺点是大结果集处理效率低，可能超出Lambda运行时长/内存限制。

方案3：使用CTAS查询指定输出（适合批量导出场景） #

如果是定期批量导出数据的场景，可以使用CREATE TABLE AS SELECT(CTAS)语法，建表时指定外部表的S3存储路径、输出格式，可自定义文件前缀，适合大批量数据的导出场景。

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