如何通过AWS DocumentDB变更流将TTL过期文档压缩上传至S3冷存储

方案一：定时批量扫描归档（最直接可靠） #

适用场景 #

适合对实时性要求不高、每日批量处理的场景，能确保拿到完整的过期文档内容。

实现步骤 #

• 确认TTL字段配置：确保目标集合已基于自定义TTL字段（如expireAt）创建TTL索引，命令示例：

  db.your_collection.createIndex({ expireAt: 1 }, { expireAfterSeconds: 0 })
  

• 编写归档脚本：用Python/Node.js等实现以下逻辑：
  1. 连接DocumentDB集群（强制SSL连接），查询所有已过期的文档：

     from pymongo import MongoClient
     import datetime
     
     client = MongoClient("mongodb://username:password@docdb-cluster-endpoint:27017/?ssl=true&ssl_ca_certs=rds-combined-ca-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred")
     db = client.your_database
     coll = db.your_collection
     
     cutoff_time = datetime.datetime.now(datetime.timezone.utc)
     expired_docs = list(coll.find({ "expireAt": { "$lt": cutoff_time } }))
     

  2. 将文档序列化为JSON并压缩：

     import json
     import gzip
     
     docs_json = json.dumps(expired_docs, default=str).encode('utf-8')
     with gzip.open(f'expired_docs_{cutoff_time.strftime("%Y%m%d")}.json.gz', 'wb') as f:
         f.write(docs_json)
     

  3. 上传压缩文件至S3：

     import boto3
     
     s3 = boto3.client('s3')
     s3.upload_file(
         f'expired_docs_{cutoff_time.strftime("%Y%m%d")}.json.gz',
         'your-archive-bucket',
         f'documentdb-archive/{cutoff_time.strftime("%Y%m%d")}/expired_docs.json.gz'
     )
     

  4. （可选）备份完成后手动删除已归档文档，避免重复处理：

     coll.delete_many({ "expireAt": { "$lt": cutoff_time } })
     

• 定时调度：用Amazon EventBridge（原CloudWatch Events）触发Lambda执行脚本，或用ECS/Fargate定时运行容器化任务，设置每日执行一次。

方案二：基于变更流的实时捕获+每日归档 #

适用场景 #

需要实时跟踪TTL删除事件，确保不遗漏任何过期文档（注意：变更流的delete事件仅返回文档主键，需提前预存完整内容）

实现步骤 #

• 预存文档快照：在文档插入/更新时，同步将完整内容存入临时归档集合（如your_collection_archive），或利用副本集读取副本数据作为快照。
• 监听变更流捕获删除事件：编写Lambda或常驻服务过滤delete类型的变更事件，从归档集合中获取完整文档并存入S3临时目录：

  change_stream = coll.watch([{
      "$match": { "operationType": "delete" }
  }])
  
  coll_archive = db.your_collection_archive
  s3 = boto3.client('s3')
  
  for change in change_stream:
      doc_id = change['documentKey']['_id']
      archived_doc = coll_archive.find_one({ "_id": doc_id })
      if archived_doc:
          s3.put_object(
              Bucket='your-archive-bucket',
              Key=f'pending-archive/{str(doc_id)}.json',
              Body=json.dumps(archived_doc, default=str).encode('utf-8')
          )
          coll_archive.delete_one({ "_id": doc_id })
  

• 每日批量压缩归档：用定时任务扫描S3的pending-archive目录，将所有文件压缩为统一包，移动至正式归档目录并清理临时文件。

关键注意事项 #

• 连接安全：DocumentDB强制SSL连接，需使用AWS提供的rds-combined-ca-bundle.pem证书
• 权限配置：执行任务的IAM角色需具备DocumentDB读权限、S3读写权限
• 数据去重：定时扫描方案可通过添加archived: boolean标记字段，或归档后删除文档避免重复处理
• 性能优化：大量文档归档时采用流式压缩，避免内存占用过高
• 监控告警：用CloudWatch监控任务执行状态、归档文档数量，设置异常告警

内容的提问来源于stack exchange，提问作者Manoj K J