Kinesis Firehose Lambda转换 [英] Kinesis Firehose lambda transformation
本文介绍了Kinesis Firehose Lambda转换的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
作为Kinesis firehose记录转换的一部分,我具有以下lambda函数,该转换将msgpack记录从kinesis输入流转换为json.
I have the following lambda function as part of Kinesis firehose record transformation which transforms msgpack record from the kinesis input stream to json.
Lambda运行时:python 3.6
Lambda Runtime: python 3.6
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)
# Do custom processing on the payload here
output_record = {
'recordId': record['recordId'],
'result': 'Ok',
'data': json.dumps(payload, ensure_ascii=False).encode('utf8')
}
output.append(output_record)
print('Successfully processed {} records.'.format(len(event['records'])))
return {'records': output}
但是lambda抛出以下错误:
But lambda throwing the following error:
An error occurred during JSON serialization of response: b'
{
"id": "d23fd47f-3a62-4383-bcb3-abdb913ea572",
"timestamp": 1526358140730,
"message": "Hello World"
}
' is not JSON serializable
Traceback (most recent call last):
File "/var/lang/lib/python3.6/json/__init__.py", line 238, in dumps
**kw).encode(obj)
File "/var/lang/lib/python3.6/json/encoder.py", line 199, in encode
chunks = self.iterencode(o, _one_shot=True)
File "/var/lang/lib/python3.6/json/encoder.py", line 257, in iterencode
return _iterencode(o, 0)
File "/var/runtime/awslambda/bootstrap.py", line 110, in
decimal_serializer
raise TypeError(repr(o) + " is not JSON serializable")
我做错什么了吗?
推荐答案
我能够解决此问题.
这是对我有用的代码.
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)
# Do custom processing on the payload here
output_record = {
'recordId': record['recordId'],
'result': 'Ok',
'data': base64.b64encode(json.dumps(payload).encode('utf-8') + b'\n').decode('utf-8')
}
output.append(output_record)
print('Successfully processed {} records.'.format(len(event['records'])))
return {'records': output}
这篇关于Kinesis Firehose Lambda转换的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文