使用 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) => {
/* Process the list of records and transform them */
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 解决方案如下所示：
# Not pretty, but it works! Successfully adds new line to record.
# newImage comes from the DynamoDB Stream as a Python dictionary object,
# I convert it to a string before running the code below.
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')