如何向DynamoDB导入JSON数据并跳过已有记录、创建新插入项流？

核心思路：基于主键去重+条件写入，结合DynamoDB Streams自动捕获新增条目 #

方案1：条件写入（推荐，无竞态问题）

这是最直接可靠的方式，利用DynamoDB的条件表达式确保只有新条目才会被插入：

• 先给每个业务条目定义全局唯一的主键（比如用业务ID作为Partition Key，复杂场景可搭配Sort Key）
• 导入时使用PutItem或BatchWriteItem操作，给每个写入请求加上ConditionExpression，判断主键对应的条目是否不存在：

  # 单条写入示例（Python）
  table.put_item(
      Item={"id": "unique_biz_id_001", "name": "example", ...},
      ConditionExpression='attribute_not_exists(id)'
  )
  

• 重复条目会因条件不满足被DynamoDB自动拒绝，不会产生重复数据
• 只要开启了DynamoDB Streams，所有成功插入的新条目都会自动生成流记录，无需额外配置

方案2：先查后写（适合小批量低并发场景）

如果数据量不大且并发度低，可以采用先查询再写入的方式：

• 对每个条目，用主键调用GetItem查询是否已存在于表中
• 确认不存在后再执行PutItem插入操作
• 注意：高并发场景下可能出现竞态问题（两个进程同时查到条目不存在，随后都执行插入），因此这种方式仅适合小规模导入

方案3：批量预处理+S3导入（适合大规模数据）

如果是超大批量的JSON文件导入，可以先做预处理再用官方工具导入：

1. 将原始JSON文件上传到S3存储桶
2. 用Lambda或Glue Job批量提取所有条目的主键，查询DynamoDB中已存在的主键集合
3. 过滤掉已存在的条目，生成仅包含新条目的JSON文件
4. 使用DynamoDB的「Import from S3」工具导入处理后的文件

• 这种方式效率更高，避免了逐条写入的开销，同时能有效保证去重效果

关键注意事项 #

• 主键唯一性：去重的核心前提是每个条目拥有全局唯一的主键，这是整个方案的基础
• 流配置：开启DynamoDB Streams时，选择NEW_IMAGE视图类型，确保流能捕获完整的新条目数据
• 批量操作细节：使用BatchWriteItem时，要给每个PutRequest单独设置ConditionExpression，示例：

  # 批量写入的条件配置示例
  request_items = {
      "YourTableName": [
          {
              "PutRequest": {
                  "Item": {"id": "biz_id_001", ...},
                  "ConditionExpression": "attribute_not_exists(id)"
              }
          },
          {
              "PutRequest": {
                  "Item": {"id": "biz_id_002", ...},
                  "ConditionExpression": "attribute_not_exists(id)"
              }
          }
      ]
  }
  dynamodb_client.batch_write_item(RequestItems=request_items)
  

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