AWS Lambda 函数 - Kinesis 数据流处理

Question

我正在编写一个 AWS Lambda 函数以从 Kinesis 数据流中使用，并在一些处理后写入 DynamoDB 文件。示例流如下：

{'DateTime': Timestamp('2021-04-29 10:30:00-0400', tz='America/New_York'), 'ticker': 'MSFT', 'Open': 250.9600067138672, 'Close': 249.1999969482422 }
{'DateTime': Timestamp('2021-04-29 11:30:00-0400', tz='America/New_York'), 'ticker': 'MSFT', 'Open': 249.2100067138672, 'Close': 250.0399932861328}

我的 Lambda 函数代码如下：

def lambda_handler(event, context):
    dynamodb_res = boto3.resource('dynamodb', region_name='us-east-1')

    eventText =json.dumps(event)
    dict1 = json.loads(eventText)
    pprint(dict1, sort_dicts=False)
    table = dynamodb_res.Table('Table1')
    for i in dict1:
      response = table.put_item(Item=i)

我不断收到以下错误。关于如何纠正这个问题的任何想法..

[ERROR] ParamValidationError: Parameter validation failed:
Invalid type for parameter Item, value: Records, type: <class 'str'>, valid types: <class 'dict'>

Answer 1

1. 查看您的 lambda 接收到的 event 对象。已经是dict，所以有2个操作没有实际意义：

   eventText =json.dumps(event)
   dict1 = json.loads(eventText)
   

2. for i in dict1 没有做你期望它做的事情。在 Python 中，当您遍历 dict 时，每次下一次迭代都会返回下一个键，在本例中为 str：

   test = {"a": 1, "b": 2}
   for key in test:
       print(f"Current key: {key}, corresponding value: {test[key]}")
   >> Current key: a, corresponding value: 1
   >> Current key: b, corresponding value: 2
   

   所以你可能打算做的是：

   for i in event["Records"]:
       response = table.put_item(Item=i)