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

const AWS = require('aws-sdk');
const fs = require('fs');

// 创建DynamoDB实例
const dynamodb = new AWS.DynamoDB.DocumentClient();

// 读取流数据并批量写入DynamoDB
async function uploadDataToDynamoDB() {
  const stream = fs.createReadStream('data.txt'); // 替换为你的数据流文件路径
  const tableName = 'YourTableName'; // 替换为你的DynamoDB表名

  let items = [];
  let count = 0;

  // 逐行读取流数据
  for await (const chunk of stream) {
    const lines = chunk.toString().split('\n');

    // 解析每行数据并构建DynamoDB写入请求
    for (const line of lines) {
      const data = JSON.parse(line);
      const params = {
        RequestItems: {
          [tableName]: [
            {
              PutRequest: {
                Item: data,
              },
            },
          ],
        },
      };

      items.push(params);
      count++;

      // 每1000条数据进行批量写入
      if (count % 1000 === 0) {
        await batchWrite(items);
        items = [];
      }
    }
  }

  // 处理剩余的数据
  if (items.length > 0) {
    await batchWrite(items);
  }

  console.log('数据上传完成');
}

// 执行批量写入操作
async function batchWrite(items) {
  const params = {
    RequestItems: {},
  };

  params.RequestItems[tableName] = items;

  try {
    await dynamodb.batchWrite(params).promise();
    console.log('批量写入成功');
  } catch (error) {
    console.error('批量写入失败', error);
  }
}

// 调用函数上传数据
uploadDataToDynamoDB();