跨账户CloudTrail日志经CloudWatch/Kinesis流转适配Athena的Lambda修改咨询

现有代码的核心问题 #

• 字段格式不符合Firehose Lambda契约要求：返回记录的data字段必须是Base64编码后的字符串，原代码直接将Python字典赋值给data，最终写入S3的内容是无法解析的二进制乱码
• 输出结构不符合Athena最优解析格式：原逻辑强行套外层Records数组包装，这种嵌套结构会导致Athena需要定义复杂的数组类型、查询时手动展开，性能极差
• 大小判断逻辑完全失效：原代码用len(result)计算大小，Python字典的len()返回的是键值对数量，不是序列化后的实际字节数，根本无法判断是否超过Firehose单条记录6MB的上限
• 结构认知偏差：原代码参考的Records数组格式是Kinesis事件投递到Lambda的入参结构，不是S3存储的输出格式

核心逻辑调整（对应elif data['messageType'] == 'DATA_MESSAGE'分支） #

适配Athena的最优存储格式为NDJSON（每行一个独立的合法JSON对象），这种格式下Athena可以直接用原生JSON SerDe解析，不需要额外嵌套结构，查询效率最高。

原错误代码片段 #

elif data['messageType'] == 'DATA_MESSAGE':
    result = {}
    result["Records"] = {}
    events = []
    for e in data['logEvents']:
        events.append(e["message"])
    result["Records"] = events
    print(result)
    
    if len(result) <= 6000000:
        yield {
            'data': result,
            'result': 'Ok',
            'recordId': recId
        }
    else:
        yield {
            'result': 'ProcessingFailed',
            'recordId': recId
        }

修改后可正常运行的代码片段 #

elif data['messageType'] == 'DATA_MESSAGE':
    # 收集同批次内所有合法的CloudTrail日志行
    valid_log_lines = []
    for log_event in data['logEvents']:
        log_content = log_event.get('message', '').strip()
        # 跳过空行，避免Athena解析报错
        if not log_content:
            continue
        # 校验日志本身是合法JSON，过滤脏数据
        try:
            json.loads(log_content)
            valid_log_lines.append(log_content)
        except json.JSONDecodeError:
            print(f"跳过非法日志: {log_content}")
            continue
    
    # 拼接为NDJSON格式（每行一个JSON对象）
    output_content = '\n'.join(valid_log_lines)
    # 计算实际序列化后的字节大小，判断是否超过Firehose 6MB上限
    content_size = len(output_content.encode('utf-8'))
    
    if content_size <= 6000000:
        # 必须对输出内容做Base64编码，符合Firehose Lambda的返回契约
        encoded_data = base64.b64encode(output_content.encode('utf-8')).decode('utf-8')
        yield {
            'data': encoded_data,
            'result': 'Ok',
            'recordId': recId
        }
    else:
        # 超过大小阈值的记录触发重投逻辑
        yield {
            'result': 'ProcessingFailed',
            'recordId': recId
        }

配套调整的关联逻辑 #

原lambda_handler函数中计算投影大小的逻辑，要对应匹配修改：

# 原逻辑兼容：rec['data']已经是Base64字符串，直接取长度即可，不需要额外处理
projectedSize += len(rec['data']) + len(rec['recordId'])

可选：如果坚持使用外层Records结构的实现方式 #

如果确实需要保留外层{"Records": [...]}的包装结构（不推荐，Athena查询效率低），必须按以下步骤实现，不能直接塞字典：

1. 把Records数组内容序列化为合法JSON字符串
2. 对整个JSON字符串做Base64编码
3. 编码后的内容作为data字段返回
   参考实现：

# 仅作演示，不推荐生产使用
output_obj = {"Records": valid_log_lines}
output_str = json.dumps(output_obj)
encoded_data = base64.b64encode(output_str.encode('utf-8')).decode('utf-8')

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