查询和更新dynamoDb的最有效方法

Question

我有一个dynamo DB表，用于存储失败的请求，稍后另一个lambda将希望读取请求并重新处理它们

此时，我正在使用typescript CDK创建这样的表

const myTable = new dynamodb.Table(this, "my-table", {
      tableName: "my-table-name",
      partitionKey: { name: "file_id", type: dynamodb.AttributeType.STRING },
    });

我在python lambda中向表中发送数据，如下所示

dynamodb = boto3.resource("dynamodb", region_name=region)
my_table = dynamodb.Table("my-table-name")

failedRecord = {
        "file_id": str(file_id),
        "processed": "false",
        "payload": str(payload),
    }

    my_table.put_item(Item=failedRecord)

现在，我要从另一个lambda中执行的操作是，对于表中所有processed=false的条目，我要读取它们，对它们执行一些操作，然后更新它们的processed=true

我是否需要在这里添加二级索引以提高效率。举个例子来说明如何做到这一点会很好

谢谢

Answer 1

考虑创建一个全局二级索引，其中包含仅< EEM>未处理的项。您可以通过添加/删除GSI主键来添加/删除GSI中的项。例如，考虑下面的表结构：

请注意，只有file_id3和4定义了GSIPK。GSI在逻辑上如下所示：

DynamoDB只会将项投影到该项上存在GSIPK的索引中。您的lambda可以从GSI读取数据，做一些工作，将processed属性设置为true，并删除GSIPK值。这将有效地从二级索引中删除该项

对DynamoDB进行的update调用如下所示：

 const params = {
    TableName: YOUR_TABLE_NAME_HERE,
    Key: {
      PK: FILE_ID_HERE
    },
    UpdateExpression: "SET #processed = :true REMOVE #gsipk",
    ExpressionAttributeNames: {
      "#processed": "processed",
      "#gsi1pk": "GSIPK",
    },
    ExpressionAttributeValues: {
      ":true": true
    }
  };

  ddbClient.update(params);

Answer 2

假设您的filenote_id已经是唯一的（假定您已经将其设置为分区键），使用您共享的记录格式和表模式GSI，而不添加排序键，则不会有任何区别

您可以考虑的另一种方法是为所讨论的表启用DynamoDB Stream，并将其设置为trigger of the second Lambda Function。 使用这种方法，您基本上可以捕获表上的所有活动，并且在您的逻辑中，您可以过滤掉所有不是INSERT的事件，并按照您自己的速度处理您感兴趣的事件

这样可以避免完全查询表