Node Lambda中通过aws s3.getObject.createReadStream()读取CSV行的问题

问题根因 #

导致流解析逻辑无法触发、Lambda提前结束的核心原因有3个：

1. Promise构造器的执行函数传入了async关键字，属于错误的Promise写法，一旦内部抛出异常，外层Promise无法捕获，会造成任务卡死或者Lambda提前退出。
2. 流的error事件仅做了日志打印，没有调用Promise的reject方法，一旦S3读取或CSV解析出错，Promise会一直处于pending状态，Lambda会因为超时或运行时回收直接终止。
3. Lambda handler调用readFileStreamRowByRow方法时未添加await，Lambda判断异步任务已完成，直接终止执行流程，不会等待流的end事件触发。

修正后的实现代码 #

S3Service 优化代码 #

const AWS = require('aws-sdk');
const csv = require('fast-csv');

class S3Service {
  constructor(s3 = new AWS.S3()) {
    this.s3 = s3;
  }

  // createReadStream为同步返回方法，不需要async修饰
  _createReadStream(bucket, key) {
    console.log('getting a stream');
    return this.s3.getObject({ Bucket: bucket, Key: key }).createReadStream();
  }

  readFileStreamRowByRow(bucket, key) {
    // 移除Promise执行函数的async修饰符
    return new Promise((resolve, reject) => {
      const rows = [];
      console.log('inside readFileStreamRowByRow');
      const stream = this._createReadStream(bucket, key);
      console.log('here is the stream', stream);
      
      // 监听原始S3流的错误，避免S3访问异常未捕获
      stream.on('error', error => {
        console.error('S3 read error:', error);
        reject(error);
      });

      stream.pipe(csv.parse({ headers: true }))
        .on('error', error => {
          console.error('CSV parse error:', error);
          reject(error);
        })
        .on('data', row => rows.push(row))
        .on('end', () => {
          console.log(`Parsed total ${rows.length} rows`);
          resolve(rows);
        });
    });
  }
}

Lambda Handler 正确调用示例 #

exports.handler = async (event) => {
  const s3Service = new S3Service();
  // 从S3触发事件中提取存储桶和文件key，可根据实际场景替换为固定值
  const bucket = event.Records[0].s3.bucket.name;
  const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));

  try {
    // 必须添加await，等待CSV全量解析完成再执行后续逻辑
    const rows = await s3Service.readFileStreamRowByRow(bucket, key);
    // 逐行执行数据库操作
    for (const row of rows) {
      // 此处替换为你的数据库操作逻辑，建议批量写入提升性能
      console.log('Processing row:', row);
      // await db.insert(row)
    }
    return {
      statusCode: 200,
      body: `Successfully processed ${rows.length} rows`
    };
  } catch (err) {
    console.error('Processing failed:', err);
    throw err;
  }
};

额外注意事项 #

• 若处理的CSV文件较大，不要将所有行存入内存数组，可直接在data事件回调中执行数据库操作，避免内存溢出，同时加快处理速度。
• 确认Lambda执行角色具备目标S3存储桶的s3:GetObject权限，以及对应数据库的访问权限；如果数据库部署在VPC内，需要为Lambda配置对应VPC访问权限，访问公网资源需要配置NAT网关。
• 根据CSV文件大小调整Lambda的超时时间，预留足够的处理时长，避免任务未完成被Lambda强制终止。

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