Firehose转换输出的格式化

Question

我正在使用AWS Kinesis Firehose进行自定义数据转换。拉姆达的用Python编写的3.6和返回字符串如下所示：

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

这LAMBDA是非常高兴，并记录看起来像上面的只是其送回流水之前的输出。然而，流水的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

我发现了使用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 2

如果你的数据是一个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'的属性需要一个普通的'xxx'字符串。