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 Logs会显示错误:

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')
}