查询和更新 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 中对表中的所有条目进行处理 = false 我想读取它们，对它们做一些事情，然后更新它们的处理 = true。

是否需要在此处添加二级索引才能提高效率。一个如何做到这一点的例子会很棒。

谢谢

Answer 1

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

请注意，只有 file_id 3 和 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 的事件，并按照自己的节奏处理您感兴趣的事件。

这样您就可以完全避免查询表。