使用Lambda捕获Athena查询执行详情写入S3失败如何排查？

数据老陈

2026-10-6

故障原因

代码缩进错误：你提供的代码中，调用Athena接口、写入S3的所有业务逻辑都没有缩进，不属于lambda_handler函数的内部代码。Lambda函数触发时只会执行lambda_handler函数内的逻辑，这部分外部代码仅会在冷启动阶段执行一次，后续触发都不会运行，自然不会写入S3。
占位符未正确替换：代码中<my_region>、<s3_bucket_path>两个占位符如果没有替换为实际的AWS区域、S3桶名，会直接导致接口调用失败。注意s3.Object的第一个参数只能是纯桶名，不能带s3://前缀或者路径后缀。
IAM权限缺失：Lambda绑定的执行角色缺少必要权限，至少需要两个权限：athena:GetQueryExecution（用于调用Athena接口查询查询详情）、目标S3桶的s3:PutObject权限（用于写入文件）。
事件触发配置错误：CloudWatch事件规则的事件模式如果没有正确匹配Athena查询状态变更事件，Lambda根本不会被触发；另外EventBridge没有调用Lambda的权限也会导致触发失败。
代码鲁棒性不足：如果传入的事件结构缺少detail或者QueryExecutionId字段，直接取字段会抛出异常导致执行中断；另外你使用str()转换查询结果写入S3，存储的是Python对象的字符串格式，虽然不影响写入成功，但也可能是你误以为没写入的原因。

解决方案

首先修复代码缩进：将lambda_handler函数外的所有业务逻辑（从创建Athena客户端到return 0的全部代码）缩进4格，放到函数内部。
替换所有占位符：将<my_region>替换为Athena实际部署的区域，将<s3_bucket_path>替换为目标S3桶的纯桶名。
补充Lambda执行角色权限，参考权限策略示例：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "athena:GetQueryExecution",
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::<你的桶名>/query_statistics_json/*"
        }
    ]
}

验证CloudWatch事件规则：确认事件模式匹配Athena查询状态变更，参考事件模式如下：

{
  "source": ["aws.athena"],
  "detail-type": ["Athena Query State Change"]
}

同时确认事件规则的Lambda目标已经配置了调用权限，控制台创建规则时会自动生成该权限，CLI创建需手动配置。

优化代码鲁棒性，参考修正后的代码：

import json
import boto3
from botocore.config import Config

my_config = Config(region_name = '<替换为实际区域>')
print('Loading function')

def lambda_handler(event, context):
    print("Received event: " + json.dumps(event))
    # 先判断字段是否存在，避免报错
    if 'detail' not in event or 'QueryExecutionId' not in event['detail']:
        print("无效事件，缺少QueryExecutionId")
        return 1
    query_execution_id = event['detail']['QueryExecutionId']
    print("QueryID: " + query_execution_id)
    
    try:
        # get query statistics
        client = boto3.client('athena', config=my_config)
        queries = client.get_query_execution(QueryExecutionId=query_execution_id)
        del queries['QueryExecution']['Status']

        # 保存为标准JSON格式到S3
        s3 = boto3.resource('s3')
        object = s3.Object('<替换为实际桶名>', f'query_statistics_json/{query_execution_id}')
        object.put(Body=json.dumps(queries['QueryExecution']))
        print("写入S3成功")
        return 0
    except Exception as e:
        print(f"执行失败: {str(e)}")
        return 1

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