Node.js SDK如何简洁实现DynamoDB batchWriteItem批量写入

问题背景 #

我有一个需要对DynamoDB执行批量写入操作的业务场景，参考同类场景的公开实现方案，对代码做少量清理调整后完成了开发，功能运行符合预期。
原有实现代码如下：

const _ = require('lodash');

// helper methods

async function asyncForEach(array, cb) {
  await Promise.all(array.map(async (item) => {
    await cb(item, array);
  }));
}

function to(promise) {
  return promise.then((data) => [null, data])
    .catch((err) => [err]);
}
const call = function (params) {
  return dynamoDb.batchWriteItem(params).promise();
};
async function batchWrite25(arrayOf25, tableName) {
  // 单次批量写入最大支持25条
  const itemsArray = [];
  _.forEach(arrayOf25, (item) => {
    itemsArray.push({
      PutRequest: {
        Item: item,
      },
    });
  });
    const params = {
        RequestItems: {
        [tableName]: itemsArray,
        },
    };

  await to(call(params));
}
async function batchWrite(itemArray, tableName) {
  let mainIndex = 0;
  let subIndex = 0;
  let arrayOf25 = [];
  const arrayLength = itemArray.length;
  await asyncForEach(itemArray, async (item) => {
    arrayOf25.push(item);
    subIndex += 1;
    mainIndex += 1;

    // 单次批量写入最大支持25条
    if (subIndex % 25 === 0 || mainIndex === arrayLength) {
      await to(batchWrite25(arrayOf25, tableName));
      subIndex = 0; // 重置计数
      arrayOf25 = [];
    }
  });
}

module.exports = {
  batchWrite,
};

当前代码自定义了较多辅助方法，结构偏复杂，希望找到更简洁的实现方式，无需使用call、asyncForEach、to这几个自定义辅助方法，就能实现相同的批量写入功能。

优化后实现 #

完全可以移除这三个无必要的自定义辅助方法，依托JavaScript原生语法和Lodash内置工具方法，就能实现更简洁、可读性更高的版本，同时保留原有功能特性：

const _ = require('lodash');

/**
 * DynamoDB批量写入，自动按接口限制分片提交
 * @param {Array} itemArray 待写入的数据条目数组
 * @param {string} tableName 目标DynamoDB表名
 * @returns {Promise<Array>} 每批次写入结果集合，单条格式为[错误对象, 响应结果]，无错误时错误对象为null
 */
async function batchWrite(itemArray, tableName) {
  // *DynamoDB原生batchWriteItem接口单次调用最多支持25条写入请求*，直接用lodash内置chunk方法分片
  const chunks = _.chunk(itemArray, 25);
  const batchResults = [];

  // 用原生for...of串行遍历分片，避免并发触发写入限流
  for (const singleChunk of chunks) {
    // 直接构造请求参数
    const requestParams = {
      RequestItems: {
        [tableName]: singleChunk.map(item => ({
          PutRequest: { Item: item }
        }))
      }
    };

    // 用原生try/catch做错误捕获，替代自定义to方法
    try {
      const res = await dynamoDb.batchWriteItem(requestParams).promise();
      batchResults.push([null, res]);
    } catch (err) {
      batchResults.push([err]);
    }
  }

  return batchResults;
}

module.exports = { batchWrite };

优化说明 #

• 移除asyncForEach：不需要自定义遍历方法，原生for...of循环即可满足遍历需求，同时修正了原实现基于Promise.all并发发起所有批次请求、容易触发DynamoDB限流的问题
• 移除to：用JavaScript原生try/catch语法做Promise错误捕获，语义更清晰，无额外封装成本
• 移除call：单层封装SDK调用没有实际复用价值，直接在逻辑内调用batchWriteItem方法即可
• 移除手动维护索引分片的逻辑：直接使用Lodash内置_.chunk方法完成数组切分，代码量更少，也避免手动计数容易出现的边界判断bug

补充提示：如果业务需要保证写入成功率，可以在每批次请求拿到结果后，判断返回值中的UnprocessedItems字段，对未写入成功的条目做重试即可。

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