如何基于排序键模式批量删除DynamoDB中的旧数据？

我有一个结构如下的DynamoDB表：

需要执行批量删除操作，满足条件：

• pk='abv'
• sk格式为[1-30之间的整数]#[YYYY-MM-DD]，且日期部分早于当前日期（示例中当前日期为2023-12-12）

最终只保留日期为2023-12-12的条目：

现有实现使用了scan()操作，但希望避免这种低效方式，求更优的批量删除方案。

现有代码：

from datetime import datetime, timedelta
from typing import Dict, List
class Dynamodb:

    def batch_delete_old_data(self, pk: str):
        try:
            # Calculate the date to keep (e.g., today's date)
            date_to_keep = datetime.now().strftime('%Y-%m-%d')

            # Scan for all items with the specified pk
            response = self._table.scan(
                FilterExpression=Key('pk').eq(pk)
            )

            items_to_delete = [{'pk': item['pk'], 'sk': item['sk']} for item in response.get('Items', [])
                               if self.extract_date_part(item['sk']) < date_to_keep]

            with self._table.batch_writer() as batch:
                for item in items_to_delete:
                    batch.delete_item(Key=item)

            return {"message": "Old data cleanup successful"}

        except Exception as e:
            # Handle errors appropriately
            raise Exception(f"Error: {str(e)}")

    @staticmethod
    def extract_date_part(sk: str) -> str:
        # Extract the date part from the sk, assuming format "prefix#date"
        return sk.split('#')[-1] if '#' in sk else sk

由于你的表中pk是分区键，sk是排序键，完全可以通过Query操作精准筛选需要删除的条目，避免全表扫描的低效问题。核心思路是利用sk的格式特性，构造排序键的范围条件，结合分区键过滤，快速定位目标数据。

具体实现步骤 #

1. 构造排序键的范围条件：
   由于sk的格式是数字#日期，而日期是YYYY-MM-DD格式（字符串比较与日期顺序一致），我们可以构造两个边界值：

   • 起始边界：1#1970-01-01（最小可能的sk值）
   • 结束边界：30#[当前日期的前一天]（因为要删除早于当前日期的条目，所有数字#[早于当前日期]的sk都符合条件）

2. 使用Query批量获取目标条目：
   通过Query操作，指定pk='abv'，同时sk的范围是BETWEEN '1#1970-01-01' AND '30#{yesterday}'，这样就能精准获取所有需要删除的条目，无需全表扫描。

3. 使用BatchWriter批量删除：
   依然保留batch_writer()来高效执行批量删除，注意处理Query的分页（如果结果超过1MB会自动分页，需要循环获取所有结果）。

优化后的代码实现 #

from datetime import datetime, timedelta
from typing import Dict, List
from boto3.dynamodb.conditions import Key

class Dynamodb:
    def __init__(self, table):
        self._table = table

    def batch_delete_old_data(self, pk: str):
        try:
            # 计算当前日期和前一天日期（用于构造sk的结束边界）
            today = datetime.now().date()
            yesterday = today - timedelta(days=1)
            date_threshold = yesterday.strftime('%Y-%m-%d')
            
            # 构造sk的范围边界
            sk_start = '1#1970-01-01'
            sk_end = f'30#{date_threshold}'

            items_to_delete = []
            last_evaluated_key = None

            # 循环处理Query的分页结果
            while True:
                query_kwargs = {
                    'KeyConditionExpression': Key('pk').eq(pk) & Key('sk').between(sk_start, sk_end)
                }
                if last_evaluated_key:
                    query_kwargs['ExclusiveStartKey'] = last_evaluated_key
                
                response = self._table.query(**query_kwargs)
                items_to_delete.extend([{'pk': item['pk'], 'sk': item['sk']} for item in response.get('Items', [])])
                
                last_evaluated_key = response.get('LastEvaluatedKey')
                if not last_evaluated_key:
                    break

            # 批量删除条目
            if items_to_delete:
                with self._table.batch_writer() as batch:
                    for item in items_to_delete:
                        batch.delete_item(Key=item)

            return {"message": f"成功删除{len(items_to_delete)}条旧数据"}

        except Exception as e:
            raise Exception(f"清理旧数据失败: {str(e)}")

关键优化点说明 #

• 替换Scan为Query：Query操作直接针对分区键pk和排序键sk的范围进行筛选，性能远高于Scan，尤其是数据量较大时，能大幅减少读取的数据量和耗时。
• 处理分页：DynamoDB Query结果超过1MB时会返回LastEvaluatedKey，需要循环获取所有分页数据，确保没有遗漏需要删除的条目。
• 精准的sk范围：利用sk的字符串格式特性，通过BETWEEN条件直接筛选出所有日期早于当前日期的条目，无需在客户端额外过滤。

额外注意事项 #

• 如果你的sk中的数字部分可能超过30，需要调整sk_end的前缀数字（比如如果最大是100，就改成100#{date_threshold}）。
• 批量删除时，batch_writer()会自动处理请求的拆分和重试，但如果数据量极大（比如超过10万条），可以考虑分批次执行，避免单次操作耗时过长。
• 建议在非高峰时段执行清理操作，避免影响业务流量。

内容的提问来源于stack exchange，提问作者A l w a y s S u n n y