在读取流中使用 async/await 批量上传到 DynamoDB

Question

这里有点Node新手...

我正在尝试编写一个函数，从 S3 中提取 CSV 并将项目批量写入 DynamoDB。 DynamoDB 在每批中限制为 25 个，因此我需要随时编写条目。我遇到的问题是我的await 执行数据库写入的函数只在.end() 触发，而不是在我检查时触发。

我知道我无法执行这样的操作，但我不确定如何解决？我正在使用 Node12。

谢谢。

async function populateTable(
  dataFile: bucketKey,
  tableName: string
): Promise<void> {
  const s3 = getS3Client();
  const stream = s3.getObject(dataFile).createReadStream();

  const BATCH_COUNT = 25; // Max size to write to DynamoDB
  let counter = 0;
  let datarows: any = [];
  let datarow = {};

  stream
    .pipe(parse(DATA_HEADERS))
    .on("data", async function(data: DataRow) {
      counter++;

      datarow = {
        PutRequest: {
          Item: data
        }
      };

      datarows.push(datarow);

      if (counter % BATCH_COUNT === 0) {
        console.log("before batch write " + counter); // This fires!
        await batchWriteToDynamo(datarows, tableName); // I want this function to fully execute before moving on
        console.log("after batch write " + counter); // This does not
        datarows = [];
      }
    })
    .on("end", async function() {
      await batchWriteToDynamo(datarows, tableName); // This fires!
    });
}

Answer 1

我猜这些流事件不兼容异步；您可能不得不求助于创建自己的承诺链。您可以通过以下方式执行此操作：

  let datarow = {};
  let pr = Promise.resolve();
  // ...
      if (counter % BATCH_COUNT === 0) {
        let scopedRows = datarows.slice(); // scoped shallow copy
        pr = pr.then(()=> batchWriteToDynamo(scopedrows, tableName));

  // ...
    .on("end", async function() {
      pr = pr.then(()=> batchWriteToDynamo(datarows, tableName));

这应该确保您的批量写入一次发生一个并且以正确的顺序发生。还要注意数据事件期间datarows 的浅拷贝。很确定这是必要的，因为事件和承诺将以不可预测的顺序发生。

但最终它不应该是必要的，因为datarows 在那个时候不应该再改变了，我猜。