使用Python或Node.js的解决方案
我正在使用DynamoDB Streams,并且需要将这些记录保存到S3中。我实现了一个Kinesis Firehose流以及一个Lambda函数。这可以将我的记录作为JSON字符串保存到S3中,但是,保存到S3文件中的每个记录都是内联的,也就是说,在一行中连续存在,因此我需要在添加每个记录时在其末尾添加一个新行,以便每个记录都在自己的一行上。对于我的解决方案,我最终必须进行一些base64解码/编码。
以下是我执行此操作的步骤:
- 创建Kinesis Firehose流时,请启用“使用AWS Lambda转换源记录”(选择“已启用”)。如果您已经创建了流,则仍然可以通过编辑现有流来启用此功能。
- 此时,您需要选择另一个执行此转换的Lambda函数。在我的情况下,我需要在每个记录的末尾添加一个新行,以便在文本编辑器中打开文件并查看时,每个条目都在单独的一行上。
以下是我用于第二个Lambda的Python和Node.js的测试解决方案代码:
添加新行的Python解决方案:
import json
import boto3
import base64
output = []
def lambda_handler(event, context):
for record in event['records']:
payload = base64.b64decode(record['data']).decode('utf-8')
print('payload:', payload)
row_w_newline = payload + "\n"
print('row_w_newline type:', type(row_w_newline))
row_w_newline = base64.b64encode(row_w_newline.encode('utf-8'))
output_record = {
'recordId': record['recordId'],
'result': 'Ok',
'data': row_w_newline
}
output.append(output_record)
print('Processed {} records.'.format(len(event['records'])))
return {'records': output}
添加换行的Node.js解决方案:
'use strict';
console.log('Loading function');
exports.handler = (event, context, callback) => {
const output = event.records.map((record) => {
let entry = (new Buffer(record.data, 'base64')).toString('utf8');
let result = entry + "\n"
const payload = (new Buffer(result, 'utf8')).toString('base64');
return {
recordId: record.recordId,
result: 'Ok',
data: payload,
};
});
console.log(`Processing completed. Successful records ${output.length}.`);
callback(null, { records: output });
};
以下是帮助我整理Python版本的一些好参考资料:
在上面的问题中,MrHen想要在不使用第二个Lambda的情况下完成此操作。我能够在第一个Lambda中使其工作,而不是使用Kinesis Firehose转换源记录功能。我按照以下顺序对来自DynamoDB的newImage进行了编码、解码、添加新行("\n")、编码和解码处理。可能有更简洁的方法。我选择使用第二个Lambda函数作为转换源记录功能,因为目前它对我来说似乎更加清晰。
在我的情况下,单个Lambda解决方案如下:
newImage = base64.b64encode(newImage.encode('utf-8'))
newImage = base64.b64decode(newImage).decode('utf-8')
newImage = newImage + "\n"
newImage = base64.b64encode(newImage.encode('utf-8'))
newImage = base64.b64decode(newImage).decode('utf-8')