如何通过boto3客户端利用query_execution_id获取Athena查询结果大小（无需下载CSV）

完全可以通过boto3客户端实现，无需下载CSV文件。核心是调用get_query_execution方法获取查询执行元数据，再通过S3客户端读取结果文件的大小属性。

实现步骤 #

• 初始化Athena和S3的boto3客户端
• 传入目标query_execution_id调用get_query_execution获取查询状态与输出路径
• 解析S3输出路径，调用head_object获取文件大小元数据

代码示例 #

import boto3

def get_athena_result_size(query_execution_id):
    athena = boto3.client('athena')
    s3 = boto3.client('s3')

    # 获取查询执行详情
    exec_response = athena.get_query_execution(QueryExecutionId=query_execution_id)
    exec_state = exec_response['QueryExecution']['Status']['State']
    
    if exec_state != 'SUCCEEDED':
        raise ValueError(f"Query {query_execution_id} is {exec_state}, no valid result file")
    
    # 解析S3输出路径
    output_path = exec_response['QueryExecution']['ResultConfiguration']['OutputLocation']
    bucket = output_path.split('//')[1].split('/')[0]
    object_key = '/'.join(output_path.split('//')[1].split('/')[1:])
    
    # 获取文件大小（字节）
    s3_response = s3.head_object(Bucket=bucket, Key=object_key)
    size_bytes = s3_response['ContentLength']
    
    # 可选：转换为MB单位
    size_mb = round(size_bytes / (1024 * 1024), 2)
    return {'bytes': size_bytes, 'mb': size_mb}

# 使用示例
query_id = 'your-query-execution-id'
result_size = get_athena_result_size(query_id)
print(f"结果大小：{result_size['bytes']} 字节（{result_size['mb']} MB）")

注意事项 #

• 执行代码的IAM角色需具备athena:GetQueryExecution和s3:HeadObject权限
• 仅当查询状态为SUCCEEDED时，输出路径才会存在有效文件
• ContentLength返回的是字节数，可按需转换为其他存储单位

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