如何用Boto3 API批量导出AWS SQS DLQ中5万条消息至CSV

SQS的receive_message接口确实单次最多返回10条消息，无法一次性拉取所有5万条，只能通过循环批量拉取的方式实现。下面是完整的代码示例，包含消息拉取、CSV写入，以及可选的消息删除逻辑：

代码示例 #

import boto3
import csv
import time
from botocore.exceptions import ClientError

# 初始化SQS客户端
sqs = boto3.client('sqs')

# 替换为你的死信队列URL
DLQ_QUEUE_URL = 'https://sqs.<region>.amazonaws.com/<account-id>/<dlq-queue-name>'

# CSV文件路径
CSV_FILE_PATH = 'dlq_messages_backup.csv'

def backup_dlq_messages():
    # 打开CSV文件并写入表头
    with open(CSV_FILE_PATH, 'w', newline='', encoding='utf-8') as csvfile:
        fieldnames = ['MessageId', 'Timestamp', 'Body']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        message_count = 0
        while True:
            try:
                # 长轮询拉取消息，最多10条，等待2秒（减少空请求次数）
                response = sqs.receive_message(
                    QueueUrl=DLQ_QUEUE_URL,
                    MaxNumberOfMessages=10,
                    WaitTimeSeconds=2,
                    AttributeNames=['SentTimestamp']
                )
            except ClientError as e:
                print(f"拉取消息失败: {e.response['Error']['Message']}")
                time.sleep(5)
                continue

            # 没有消息时退出循环
            if 'Messages' not in response:
                print(f"备份完成，共处理 {message_count} 条消息")
                break

            messages = response['Messages']
            # 批量处理消息
            entries = []
            for msg in messages:
                # 提取消息字段
                message_id = msg['MessageId']
                timestamp = msg['Attributes']['SentTimestamp']
                body = msg['Body']

                # 写入CSV
                writer.writerow({
                    'MessageId': message_id,
                    'Timestamp': timestamp,
                    'Body': body
                })
                message_count += 1

                # 准备批量删除的条目（如果需要删除已备份的消息）
                entries.append({
                    'Id': message_id,
                    'ReceiptHandle': msg['ReceiptHandle']
                })

            # 批量删除已备份的消息（可选，根据需求注释/取消注释）
            if entries:
                try:
                    sqs.delete_message_batch(
                        QueueUrl=DLQ_QUEUE_URL,
                        Entries=entries
                    )
                except ClientError as e:
                    print(f"批量删除消息失败: {e.response['Error']['Message']}")

            # 可选：避免API调用过于频繁，添加短暂延迟
            time.sleep(0.1)

if __name__ == "__main__":
    backup_dlq_messages()

关键说明 #

• 长轮询：设置WaitTimeSeconds=2，让SQS在有消息时立即返回，无消息时最多等待2秒再返回，减少空请求次数，提升效率。
• 批量操作：用delete_message_batch批量删除消息，比单条删除更高效，适合大量消息场景。
• 异常处理：捕获ClientError，遇到错误时重试，避免程序直接崩溃。
• 字段扩展：示例仅保存MessageId、SentTimestamp和Body，若需其他属性（如ApproximateReceiveCount），可修改AttributeNames参数和CSV表头。
• Body解析：若消息Body为JSON格式，可在写入前用json.loads(body)解析，提取所需字段后再保存。

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