我可以在单个 Lambda 函数中创建和填充 dynamodb 表吗？

Question

我的代码基本上由两部分组成，它们各自独立工作，但不能一起工作。所以我认为我有语法问题。

第一部分是创建表，第二部分是填充它。 问题是，两个部分共享表名的变量。

   import os
   import boto3
   import botocore.session

   region = os.environ.get('AWS_DEFAULT_REGION', 'us-east-2')
   session = botocore.session.get_session()
   dynamo = session.create_client('dynamodb', region_name=region) 


   s3 = boto3.client('s3')
   dynamodb = boto3.resource('dynamodb')

   def lambda_handler(event, context):

    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    obj = s3.get_object(Bucket=bucket, Key=key)

    rows = obj['Body'].read().decode("utf-8"). split ('\n')

    table = dynamodb.Table(key)
    dynamodb.create_table(
    TableName=key,
    KeySchema=[
        {
            'AttributeName': 'first',
            'KeyType': 'HASH'  #Partition key
        },
        {
            'AttributeName': 'last',
            'KeyType': 'RANGE'  #Sort key
        }
    ],
    AttributeDefinitions=[
        {
            'AttributeName': 'first',
            'AttributeType': 'S'
        },
        {
            'AttributeName': 'last',
            'AttributeType': 'S'
        },

    ],
    ProvisionedThroughput={
        'ReadCapacityUnits': 5,
        'WriteCapacityUnits': 5
        }
    )
   # Wait for the table to exist before exiting
    print('Waiting for', key, '...')
    waiter = dynamo.get_waiter('table_exists')
    waiter.wait(TableName=key)
    with table.batch_writer() as batch:
        for row in rows:

            batch.put_item(Item={

                'first':row.split(',')[0],
                'last':row.split(',')[1],
                'age':row.split(',')[2],
                'date':row.split(',')[3]

            })

只要将 CSV 放入我的 s3 存储桶，它就会作为 lambda 函数运行。

运行后，成功创建表，但不填充。结尾： “任务在 3.00 秒后超时” 几秒钟后它再次启动并返回“表已存在”，但仍为空。

如果我只运行 batch_writer 部分，只要它已经存在，它就会填充表。

Answer 1

简短的回答是，新表通常需要大约一秒钟才能激活，而Waiter.TableExists 使用default polling interval of 20 seconds，这会导致您的 lambda 函数超时。

但到底发生了什么？

在内部，Waiter.TableExists 的功能大致类似于此伪代码。 （为简单起见，我省略了错误处理和其他细节。）

function waitForTable(tableName):
    while true:
        if (dynamodb.describeTable(tableName).status == active):
            return
        else:
            sleep 20 seconds

创建表后，立即启动服务员。当服务员调用describeTable时，它看到该表还没有活动，所以它等待了20秒。您的 lambda 超时设置为 3 秒，因此 3 秒后（在服务员再次调用 describeTable 之前）您的 lambda 函数将终止。 （这就是“任务超时”消息的含义。）

然后，当您的 lambda 函数被重试时，该表现在处于活动状态，因此当您的 lambda 函数到达dynamodb.create_table(...) 调用时，DynamoDB 将响应错误，因为该表已经存在。 （因此“表已存在”错误消息。）

我该如何解决？

您可以采取多种措施来解决此问题，“最正确”的解决方案可能是全部解决。

1. 您可以将服务员的延迟时间设置为较小的数字，例如 1 秒，如下所示： waiter.wait(TableName=key, WaiterConfig={'Delay': 1})
2. 您可以增加 lambda 函数的超时时间。创建表、读取 S3 文件并将其全部写入 DynamoDB 的组合可能需要 3 秒以上。如果需要重试请求，请选择一个数字，让您的 lambda 函数有时间恢复。如果您的函数适用于只有 1-2 行的文件，但对于较大的文件失败，我建议尝试 5 秒，如果不能可靠地成功，请增加到 10 秒。如果文件可能非常大，您应该考虑使用 Lambda 以外的其他工具。
3. 假设您不担心覆盖预先存在的表中的数据，那么您应该在尝试创建表之前检查该表是否已经存在（或者尝试创建它并忽略如果出现的 ResourceAlreadyInUseException它已经存在）。请参阅How to check if a DynamoDB table exists 的另一个 SO 答案，其中解释了检查表是否存在的多种方法，包括每种方法的代码示例。