如何在不使用主键的情况下通过条件更新AWS DynamoDB表？

Got it, let's tackle this DynamoDB update problem you're working on. First off, a quick clarification: even though you don't want to rely on knowing primary keys upfront, DynamoDB still requires the primary key to update an item—so the solution is to first find all items matching your conditions (to grab their primary keys), then batch update them. Here's how to handle this properly:

General Workflow #

• Step 1: Locate all items matching your filter rules using a Scan (or a Query if you set up a suitable Global Secondary Index, GSI—way more efficient for large tables)
• Step 2: Extract the primary keys from these matching items
• Step 3: Use BatchWriteItem to update the items in batches (DynamoDB caps batch operations at 25 items per request)

Key Optimization #

If you frequently filter by attributes like gameid, create a GSI with gameid as the partition key. This lets you swap the full-table Scan for a targeted Query, which cuts down on cost and latency significantly.

Below is a concrete implementation using Python and boto3 (the most widely used AWS SDK). I'll assume your table has a primary key named user_id—adjust this to match your actual primary key schema.

Step 1: Initialize the DynamoDB Client #

import boto3
from boto3.dynamodb.conditions import Attr

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your-table-name')
target_gameid = 'xxxx'

Step 2: Fetch Matching Items #

Choose either the Scan (for tables without a GSI) or Query (with a GSI) implementation below:

Scan Implementation (No GSI)

matching_items = []
last_evaluated_key = None

# Paginate through scan results to get all matching items
while True:
    scan_params = {
        'FilterExpression': Attr('gameid').eq(target_gameid) & Attr('age').gt(30)
    }
    if last_evaluated_key:
        scan_params['ExclusiveStartKey'] = last_evaluated_key
    
    response = table.scan(**scan_params)
    matching_items.extend(response['Items'])
    last_evaluated_key = response.get('LastEvaluatedKey')
    
    if not last_evaluated_key:
        break

Query Implementation (With GSI on gameid)

If you created a GSI named gameid-index with gameid as the partition key, use this faster alternative:

matching_items = []
last_evaluated_key = None

while True:
    query_params = {
        'IndexName': 'gameid-index',
        'KeyConditionExpression': Attr('gameid').eq(target_gameid),
        'FilterExpression': Attr('age').gt(30)
    }
    if last_evaluated_key:
        query_params['ExclusiveStartKey'] = last_evaluated_key
    
    response = table.query(**query_params)
    matching_items.extend(response['Items'])
    last_evaluated_key = response.get('LastEvaluatedKey')
    
    if not last_evaluated_key:
        break

Step 3: Batch Update the Items #

This function handles batch updates, including retries for any unprocessed items (common with throttling):

def batch_update_active_status(table, items):
    batch_size = 25
    # Split items into batches of 25
    batches = [items[i:i+batch_size] for i in range(0, len(items), batch_size)]
    
    for batch in batches:
        request_items = {
            table.name: []
        }
        for item in batch:
            # Grab the primary key from the item (adjust 'user_id' to your actual key)
            item_key = {'user_id': item['user_id']}
            update_entry = {
                'UpdateRequest': {
                    'Key': item_key,
                    'UpdateExpression': 'SET active = :new_val',
                    'ExpressionAttributeValues': {':new_val': False}
                }
            }
            request_items[table.name].append(update_entry)
        
        # Submit the batch request
        response = dynamodb.batch_write_item(RequestItems=request_items)
        
        # Retry unprocessed items if needed
        while response.get('UnprocessedItems'):
            response = dynamodb.batch_write_item(RequestItems=response['UnprocessedItems'])

# Execute the update
if matching_items:
    batch_update_active_status(table, matching_items)
    print(f"Successfully updated {len(matching_items)} items.")
else:
    print("No items matched your specified conditions.")

Important Notes #

• Primary Key Requirement: DynamoDB can't update an item without its primary key—this workflow just avoids you having to know the keys upfront; we retrieve them via the scan/query.
• Throttling: For high-traffic tables, add small delays between batches or implement exponential backoff for unprocessed items to avoid hitting rate limits.
• Cost: Scans are more expensive for large tables (they read every item), so setting up a GSI is strongly recommended to reduce costs and speed up operations.

内容的提问来源于stack exchange，提问作者justdan0227