Kinesis Firehose Lambda转换报错：JSON序列化失败问题排查

我来帮你看看这个问题！你遇到的错误核心原因是Lambda返回的响应里包含了无法被JSON序列化的字节对象，咱们一步步拆解问题和解决方案：

问题根源 #

你代码里的这一行直接导致了报错：

'data': json.dumps(payload, ensure_ascii=False).encode('utf8')

• json.dumps(...) 生成的是Unicode字符串，调用.encode('utf8')后得到了bytes类型的对象
• Lambda在把你的返回值（{'records': output}）序列化成JSON返回给Firehose时，Python的JSON序列化器无法处理bytes对象，因此抛出了is not JSON serializable的错误
• 另外，Kinesis Firehose的记录转换程序要求输出的data字段必须是base64编码的字符串，你直接传入字节串也不符合这个格式要求

修正后的代码 #

只需要调整data字段的处理逻辑，把字节串转成base64编码的字符串即可：

from __future__ import print_function
import base64
import msgpack
import json

print('Loading function')

def lambda_handler(event, context):
    output = []
    for record in event['records']:
        payload = msgpack.unpackb(base64.b64decode(record['data']), raw=False)
        # 先将payload转为JSON字符串，再编码为utf-8字节，最后做base64编码并转成字符串
        json_str = json.dumps(payload, ensure_ascii=False)
        encoded_data = base64.b64encode(json_str.encode('utf8')).decode('utf8')
        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': encoded_data
        }
        output.append(output_record)
    print('Successfully processed {} records.'.format(len(event['records'])))
    return {'records': output}

关键修改点 #

• 先通过json.dumps生成JSON字符串，再编码为utf-8字节
• 使用base64.b64encode对字节进行编码，最后调用.decode('utf8')把base64编码后的bytes转成字符串
• 这样data字段就是一个标准的字符串，既符合JSON序列化要求，也满足Firehose对输出格式的规定

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