DynamoDB BatchWriteItem：提供的项目键列表包含重复项

Question

我正在尝试使用 DynamoDB 操作BatchWriteItem，其中我想将多条记录插入到 一个 表中。

这个表有一个分区键和一个排序键。

我正在使用 AWS lambda 和 Go 语言。

我得到要插入到切片中的元素。

我正在遵循这个程序。

1. 创建PutRequest 结构并为列表中的第一条记录添加属性值。

2. 我正在从这个PutRequest 创建WriteRequest

3. 我将此WriteRequest 添加到array of WriteRequests

4. 我正在创建由RequestItems 组成的BatchWriteItemInput，它基本上是一个表名的映射 和WriteRequests 的数组。

之后我打电话给BatchWriteItem，结果报错：

Provided list of item keys contains duplicates.

任何指针，为什么会发生这种情况？

Answer 1

您提供了两个或多个具有相同分区/排序键的项目。

根据BatchWriteItem 文档，您不能在同一个 BatchWriteItem 请求中对同一个项目执行多个操作。

Answer 2

使用batch_writer 代替batch_write_item：

import boto3

dynamodb = boto3.resource("dynamodb", region_name='eu-west-1')
my_table = dynamodb.Table('mirrorfm_yt_tracks')

with my_table.batch_writer(overwrite_by_pkeys=["user_id", "game_id"]) as batch:
    for item in items:
        batch.put_item(
            Item={
                'user_id': item['user_id'],
                'game_id': item['game_id'],
                'score': item['score']
            }
        )

如果你没有排序键，overwrite_by_pkeys 可以是None

这与@MiguelTrejo 的答案基本相同（谢谢！+1），但简化了

Answer 3

注意事项：此答案适用于 Python

正如@Benoit 所说，boto3 文档指出：

如果您想绕过单个批量写入请求的不重复限制为 botocore.exceptions.ClientError：调用 BatchWriteItem 操作时发生错误（ValidationException）：提供的项目键列表包含重复项。

您可以根据documentation 和source code 在批处理写入器上指定overwrite_by_pkeys=['partition_key', 'sort_key'] 以“如果与指定主键上的新请求项匹配，则在缓冲区中删除重复的请求项”。也就是说，如果组合主排序已经存在于缓冲区中，它将丢弃该请求并用新的请求替换它。

示例

假设您想要将 pandas 数据帧写入 DynamoDB 表，以下函数可能会有所帮助，

import json 
import datetime as dt 
import boto3 
import pandas as pd 
from typing import Optional

def write_dynamoDB(df:'pandas.core.frame.DataFrame', tbl:str, partition_key:Optional[str]=None, sort_key:Optional[str]=None):
    '''
       Function to write a pandas DataFrame to a DynamoDB Table through 
       batchWrite operation. In case there are any float values it handles 
       them by converting the data to a json format.  
       Arguments: 
       * df: pandas DataFrame to write to DynamoDB table. 
       * tbl: DynamoDB table name. 
       * partition_key (Optional): DynamoDB table partition key. 
       * sort_key (Optional): DynamoDB table sort key. 
    '''

    # Initialize AWS Resource 
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(tbl)

    # Check if overwrite keys were provided
    overwrite_keys = [partition_key, sort_key] if partition_key else None

    # Check if they are floats (convert to decimals instead) 
    if any([True for v in df.dtypes.values if v=='float64']):

        from decimal import Decimal

        # Save decimals with JSON
        df_json = json.loads(
                       json.dumps(df.to_dict(orient='records'),
                                  default=date_converter,
                                  allow_nan=True), 
                       parse_float=Decimal
                       )

        # Batch write 
        with table.batch_writer(overwrite_by_pkeys=overwrite_keys) as batch: 
            for element in df_json:
                batch.put_item(
                Item=element
            )

    else: # If there are no floats on data 

    # Batch writing 
        with table.batch_writer(overwrite_by_pkeys=overwrite_keys) as batch: 

            columns = df.columns

            for row in df.itertuples():
                batch.put_item(
                    Item={
                        col:row[idx+1] for idx,col in enumerate(columns)
                    }
                )

def date_converter(obj):
    if isinstance(obj, dt.datetime):
        return obj.__str__()
    elif isinstance(obj, dt.date):
        return obj.isoformat()

致电write_dynamoDB(dataframe, 'my_table', 'the_partition_key', 'the_sort_key')。