如何使用 Lambda 函数将 DynamoDB 数据转换为 Kinesis Firehose？

Question

import boto3
import boto3.dynamodb.types
from boto3.dynamodb.conditions import Key
import json

ddb_table_name = "panisar_test"

def lambda_handler(event, context):
    dynamodb = boto3.resource('dynamodb')
    kinesis = boto3.client('firehose')
   
    table = dynamodb.Table(ddb_table_name)
    
    ddb_keys_name = [a['AttributeName'] for a in table.attribute_definitions]
    
    response = None
    while True:
        if not response:
            response = table.scan()
        else:
            # Scan from where you stopped previously.
            response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
    
        for i in response["Items"]:
            # Get a dict of primary key(s).
            ddb_keys = {k: i[k] for k in i if k in ddb_keys_name}
            # Serialize Python Dictionaries into DynamoDB notation.
            ddb_data = boto3.dynamodb.types.TypeSerializer().serialize(i)["M"]
            ddb_keys = boto3.dynamodb.types.TypeSerializer().serialize(ddb_keys)["M"]
            # The record must contain "Keys" and "NewImage" attributes to be similar
            # to a DynamoDB Streams record. Additionally, you inject the name of
            # the source DynamoDB table in the record so you can use it as an index
            # for Amazon ES.
            record = {"Keys": ddb_keys, "NewImage": ddb_data, "SourceTable": ddb_table_name}
            # Convert the record to JSON.
            record = json.dumps(record)
            # Push the record to Amazon Kinesis.
            res = kinesis.put_record(
                DeliveryStreamName="panisr_test",
                Record=record
                )
            return{
             'res': res
            }
        # Stop the loop if no additional records are
        # available.
        if 'LastEvaluatedKey' not in response:
            break
        print(ddb_keys_name)

在这里，我正在尝试从 dynamoDb 中获取数据并发送到 firehose。据我了解，我们必须使用 put_record 方法将数据发送到 firehose。我一直在写 put_record 结构。

我明白了

"errorMessage": "Parameter validation failed:\nInvalid type for parameter Record, value: {\"Keys\": {\"id\": {\"S\": \"2\"}}, \"NewImage\": {\"id\": {\"S\": \"2\"}, \"name\": {\"S\": \"xyz\"}}, \"SourceTable\": \"panisar_test\"}, type: <class 'str'>, valid types: <class 'dict'>", this error. 

如果有人可以在这种情况下帮助我。基本上，我需要一种将数据发送回消防软管的方法。

Answer 1

您的记录格式不正确。基于docs，而不是：

record = json.dumps(record)

你应该有：

record = {'Data': json.dumps(record)}