如何在AWS Lambda(NodeJS)中从S3读取Zip文件并内存处理？

你的Lambda代码里entry事件从未触发，核心原因是Lambda的异步handler没有等待流式处理完成就提前退出：

• 你用bodyStream.pipe(unzipper.Parse())启动了流式解压，但这个操作是异步的，不会阻塞后续代码执行
• 紧接着你调用await Promise.all(updates)，此时updates还是空数组（因为entry事件还没来得及触发），Promise直接resolve，Lambda进程随即结束，导致流式处理的代码完全没机会运行

另外你的代码里还有个小问题：updates.push(promise)中的promise变量未定义，需要替换成你实际处理内容生成的Promise。

把流式解压的整个流程包装成一个Promise，确保Lambda等待所有解压和处理逻辑完成后再结束。修改后的完整代码如下：

const { S3Client, GetObjectCommand } = require('@aws-sdk/client-s3');
const unzipper = require('unzipper');

const s3 = new S3Client({
  region: process.env.AWS_S3_REGION
});

exports.handler = async (event, context) => {
  // 获取文件信息
  const bucket = event.Records[0].s3.bucket.name;
  const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));

  const params = {
    Bucket: bucket,
    Key: key,
  };

  let bodyStream;
  try {
    const s3File = await s3.send(new GetObjectCommand(params));
    bodyStream = s3File.Body;
  } catch (err) {
    throw new Error("获取文件失败: " + err.message);
  }

  // 包装流式处理为Promise，确保Lambda等待处理完成
  await new Promise((resolve, reject) => {
    const updates = [];
    bodyStream
      .pipe(unzipper.Parse())
      .on('entry', async (entry) => {
        console.log('开始处理压缩包内文件:', entry.path);
        try {
          const content = await entry.buffer();
          const body = content.toString();

          // 替换为你的实际处理逻辑，生成Promise
          const processPromise = Promise.resolve(); // 示例：这里写你的处理代码
          updates.push(processPromise);
        } catch (err) {
          reject(err);
        } finally {
          // 必须调用autodrain，否则会卡住stream
          entry.autodrain();
        }
      })
      .on('finish', async () => {
        try {
          await Promise.all(updates);
          console.log(`所有文件处理完成: ${bucket}:${key}`);
          resolve();
        } catch (err) {
          reject(err);
        }
      })
      .on('error', (err) => {
        reject(new Error("解压失败: " + err.message));
      });
  });
};

1. 用Promise包裹流式处理：通过监听finish事件等待整个压缩包处理完成，同时在该事件内等待所有处理Promise完成
2. 添加错误处理：监听stream的error事件，避免未捕获的异常导致Lambda静默失败
3. 调用entry.autodrain()：如果不需要保留entry的stream，必须调用该方法释放资源，否则stream会卡住无法触发finish事件

如果你觉得流式处理过于复杂，可以改用adm-zip库直接在内存中处理整个Zip文件（适合压缩包体积不大的场景）：

const { S3Client, GetObjectCommand } = require('@aws-sdk/client-s3');
const AdmZip = require('adm-zip');

const s3 = new S3Client({
  region: process.env.AWS_S3_REGION
});

exports.handler = async (event, context) => {
  const bucket = event.Records[0].s3.bucket.name;
  const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));

  try {
    const s3File = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
    // 将S3的stream转为Buffer
    const zipBuffer = await streamToBuffer(s3File.Body);
    const zip = new AdmZip(zipBuffer);

    // 获取所有压缩包内的文件
    const zipEntries = zip.getEntries();
    const updates = [];

    for (const entry of zipEntries) {
      if (!entry.isDirectory) {
        console.log('处理文件:', entry.entryName);
        const body = entry.getData().toString();
        // 替换为你的处理逻辑
        updates.push(Promise.resolve());
      }
    }

    await Promise.all(updates);
    console.log(`所有文件处理完成: ${bucket}:${key}`);
  } catch (err) {
    throw new Error("处理失败: " + err.message);
  }
};

// 辅助函数：将ReadableStream转为Buffer
function streamToBuffer(stream) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    stream.on('data', (chunk) => chunks.push(chunk));
    stream.on('end', () => resolve(Buffer.concat(chunks)));
    stream.on('error', reject);
  });
}

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