Firehose转换输出的格式设置

Question

我正在使用带有自定义数据转换的AWS Kinesis Firehose。Lambda是用Python 3.6编写的，返回的字符串如下所示：

{
    "records": [
        {
            "recordId": "...",
            "result": "Ok",
            "data": "..."
        },
        {
            "recordId": "...",
            "result": "Ok",
            "data": "..."
        },
        {
            "recordId": "...",
            "result": "Ok",
            "data": "..."
        }
    ]
}

这个Lambda非常好用，在将输出返回到Firehose之前，它会记录类似上述内容的输出。但是，Firehose的S3日志随后显示一个错误：

Invalid output structure: Please check your function and make sure the processed records contain valid result status of Dropped, Ok, or ProcessingFailed.

看着JS和Java在网络上传播的例子，我不清楚我需要做什么不同的事情；我很困惑。

Answer 1

如果您的数据是json对象，您可以尝试执行以下操作

import base64
import json
def lambda_handler(event, context):
    output = []
    for record in event['records']:
        # your own business logic.
        json_object = {...} 
        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(json.dumps(json_object).encode('utf-8')).decode('utf-8')
        }
        output.append(output_record)
    return {'records': output}

base64.b64encode函数仅适用于b'xxx‘字符串，而output_record的'data’属性需要一个普通的'xxx‘字符串。

Answer 2

我在使用Node.js时也发现了同样的错误。

在阅读文档http://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html时，我的错误不是对每条记录的数据字段进行base64编码。

我决定这样做：

{
    recordId: record.recordId,
    result: 'Ok',
    data: new Buffer(JSON.stringify(data)).toString('base64')
}

Answer 3

你可以在我的存储库中查看代码。https://github.com/hixichen/golang_lamda_decode_protobuf_firehose