DynamoDB BatchWriteItemCommand超25项批量请求的数组拆分优化问询

你当前的实现可以进一步优化，核心点在于DynamoDB的BatchWriteItemCommand单表单次最多支持25个请求项，没必要用24作为分片大小，直接用25能最大化单次请求的效率。另外，还需要处理批量写入返回的UnprocessedItems（限流或其他异常可能导致部分项写入失败），同时可以用更通用的分片逻辑简化代码。

优化后的实现方案 #

1. 通用数组分片函数 #

先封装一个复用性强的分片工具函数，把大数组拆成指定大小的子数组：

function chunkArray(arr, chunkSize) {
  const chunks = [];
  for (let i = 0; i < arr.length; i += chunkSize) {
    chunks.push(arr.slice(i, i + chunkSize));
  }
  return chunks;
}

2. 带重试逻辑的批量写入 #

DynamoDB可能返回未处理的项，需要循环重试直到所有数据都写入成功：

async function batchWriteProducts(products, dynamoDb, tableName) {
  let unprocessedItems = {};
  // 先把原始数组拆分为25个一组的块
  let productChunks = chunkArray(products, 25);

  do {
    // 如果存在未处理项，优先处理这些项
    if (Object.keys(unprocessedItems).length > 0) {
      productChunks = [unprocessedItems[tableName]];
      unprocessedItems = {};
    }

    for (const chunk of productChunks) {
      const requestItems = {
        [tableName]: chunk
      };
      const result = await dynamoDb.send(
        new BatchWriteItemCommand({ RequestItems: requestItems })
      );
      // 收集本次请求中未处理的项
      if (result.UnprocessedItems && Object.keys(result.UnprocessedItems).length > 0) {
        unprocessedItems = { ...unprocessedItems, ...result.UnprocessedItems };
      }
    }

    // 有未处理项时，短暂等待后重试（避免频繁请求触发限流）
    if (Object.keys(unprocessedItems).length > 0) {
      await new Promise(resolve => setTimeout(resolve, 1000));
    }
  } while (Object.keys(unprocessedItems).length > 0);
}

// 调用示例
await batchWriteProducts(products, dynamoDb, process.env.DYNAMODB_PRODUCTS_TABLE_NAME);

关键优化点 #

• 分片大小设为25：充分利用BatchWriteItem的单次请求上限，减少总请求次数
• 处理UnprocessedItems：避免因限流、资源不足等原因导致数据丢失，确保所有项最终写入成功
• 通用分片函数：可复用在其他需要数组分片的业务场景中
• 自动重试逻辑：无需手动介入，自动处理写入失败的情况

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