使用 Lambda 节点从 S3 上的文件在 S3 上创建一个 zip 文件

Posted 2023-03-07

【中文标题】使用 Lambda 节点从 S3 上的文件在 S3 上创建一个 zip 文件

【英文标题】：Create a zip file on S3 from files on S3 using Lambda Node

【发布时间】：2016-12-02 16:20:51

【问题描述】：

我需要创建一个 Zip 文件，其中包含位于我的 s3 存储桶中的一系列文件（视频和图像）。

目前使用下面的代码的问题是我很快就达到了 Lambda 的内存限制。

async.eachLimit(files, 10, function(file, next) 
    var params = 
        Bucket: bucket, // bucket name
        Key: file.key
    ;
    s3.getObject(params, function(err, data) 
        if (err) 
            console.log('file', file.key);
            console.log('get image files err',err, err.stack); // an error occurred
         else 
            console.log('file', file.key);
            zip.file(file.key, data.Body);
            next();
        
    );
, 
function(err) 
    if (err) 
        console.log('err', err);
     else 
        console.log('zip', zip);
        content = zip.generateNodeStream(
            type: 'nodebuffer',
            streamFiles:true
        );
        var params = 
            Bucket: bucket, // name of dest bucket
            Key: 'zipped/images.zip',
            Body: content
        ;
        s3.upload(params, function(err, data) 
            if (err) 
                console.log('upload zip to s3 err',err, err.stack); // an error occurred
             else 
                console.log(data); // successful response
            
        );
    
);

这是否可以使用 Lambda，或者我应该看看不同的 方法？

是否可以即时写入压缩的 zip 文件，从而在一定程度上消除内存问题，还是需要在压缩之前收集文件？

任何帮助将不胜感激。

【参考方案1】：

使用流可能会很棘手，因为我不确定如何将多个流通过管道传输到一个对象中。我已经使用标准文件对象做了几次。这是一个多步骤的过程，而且速度非常快。请记住，Lambda 在 Linux 中运行，因此您拥有所有 Linux 资源，包括系统 /tmp 目录。

在 /tmp 中创建一个子目录，调用“transient”或任何适合您的方式 使用 s3.getObject() 并将文件对象写入 /tmp/transient 使用 GLOB 包从 /tmp/transient 生成一个路径数组[] 循环数组和 zip.addLocalFile(array[i]); zip.writeZip('tmp/files.zip');

【讨论】：

我能看到的唯一问题是 lambda 在 tmp 目录中的存储空间限制为 500mb。在这种情况下，它也会限制最终的 zip 大小。

不确定您是否在 .zip 进程旁边运行任何文件处理，但由于数据量如此之大，您需要确保您的函数可以在 5 分钟的执行时间内完成。我的最大数据量通常是每次执行大约 20-25mg。

@Rabona 您是否设法通过 lambda 解决了这个问题？我有同样的问题。我们需要压缩一个包含大约 100Mb 图像的 1.5GB 视频文件。我们的内存用完了。我们还尝试使用具有相同图像的较小视频文件（~1gb）并获得超时。希望您可能发现了一些有用的东西，也可以帮助我们。

我们最终使用 Java 流解决方案解决了这个问题。这使我们能够绕过内存问题。

您能否分享一下您是如何使用 Java 流解决方案解决的？

【参考方案2】：

好的，我今天必须这样做，它有效。 Direct Buffer to Stream，不涉及磁盘。所以内存或磁盘限制在这里不会成为问题：

'use strict';

const AWS = require("aws-sdk");
AWS.config.update(  region: "eu-west-1"  );
const s3 = new AWS.S3(  apiVersion: '2006-03-01' );

const   _archiver = require('archiver');

//This returns us a stream.. consider it as a real pipe sending fluid to S3 bucket.. Don't forget it
const streamTo = (_bucket, _key) => 
	var stream = require('stream');
	var _pass = new stream.PassThrough();
	s3.upload(  Bucket: _bucket, Key: _key, Body: _pass , (_err, _data) =>  /*...Handle Errors Here*/  );
	return _pass;
;
      
exports.handler = async (_req, _ctx, _cb) => 
	var _keys = ['list of your file keys in s3'];
	
    var _list = await Promise.all(_keys.map(_key => new Promise((_resolve, _reject) => 
            s3.getObject(Bucket:'bucket-name', Key:_key)
                .then(_data => _resolve(  data: _data.Body, name: `$_key.split('/').pop()`  ));
        
    ))).catch(_err =>  throw new Error(_err)  );

    await new Promise((_resolve, _reject) =>  
        var _myStream = streamTo('bucket-name', 'fileName.zip');		//Now we instantiate that pipe...
        var _archive = _archiver('zip');
        _archive.on('error', err =>  throw new Error(err);  );
        
        //Your promise gets resolved when the fluid stops running... so that's when you get to close and resolve
        _myStream.on('close', _resolve);
        _myStream.on('end', _resolve);
        _myStream.on('error', _reject);
        
        _archive.pipe(_myStream);			//Pass that pipe to _archive so it can push the fluid straigh down to S3 bucket
        _list.forEach(_itm => _archive.append(_itm.data,  name: _itm.name  ) );		//And then we start adding files to it
        _archive.finalize();				//Tell is, that's all we want to add. Then when it finishes, the promise will resolve in one of those events up there
    ).catch(_err =>  throw new Error(_err)  );
    
    _cb(null,   );		//Handle response back to server
;

【讨论】：

是否可以将 zip 作为响应直接发送给“用户”？

@bobmoff，对于 Lambda，否，因为我们无法直接访问 HTTP 响应流。在常规的 NodeJs 环境中，是的，这是可能的。查看 NodeJs 文档，HTTP 请求和响应 是 Writable Streams 对象的一部分：link 就像 zlib、fs 等，但我们无法在 Lambda 中直接访问它；我们只使用回调函数与客户端通信。

我刚刚注意到您没有从 s3 流式传输对象，如果下载很多文件可能会导致内存问题，对吧？我不确定 .get 方法在游览示例中的作用，因为它在 aws-sdk 上不存在。猜你的意思是getObject。但无论如何，当您等待所有下载的对象完成时，这意味着所有这些对象必须同时存在于内存中。 Lambda 的内存有限，所以我想它必须改用 .getReadStream 并将它们附加到存档器中，还是我遗漏了什么？

@bobmoff 是的，抱歉，我们有所有 s3 操作的包装器，应该是 getObject 并且你应该传入一个返回函数。

我刚刚使用 s3.getObject().createReadStream() 进行了尝试，对于我对 3-10mb 左右的 100 个文件的测试，内存使用停止在 170mb 左右。与没有流相比，它高达 730mb。它也更快，因为它不必等待首先下载所有对象然后启动流到存储桶。

【参考方案3】：

我根据@iocoker 格式化了代码。

主条目

// index.js

'use strict';
const S3Zip = require('./s3-zip')

const params = 
  files: [
    
      fileName: '1.jpg',
      key: 'key1.JPG'
    ,
    
      fileName: '2.jpg',
      key: 'key2.JPG'
    
  ],
  zippedFileKey: 'zipped-file-key.zip'


exports.handler = async event => 
  const s3Zip = new S3Zip(params);
  await s3Zip.process();

  return 
    statusCode: 200,
    body: JSON.stringify(
      
        message: 'Zip file successfully!'
      
    )
  ;

压缩文件工具

// s3-zip.js

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

const Archiver = require('archiver');
const Stream = require('stream');

const https = require('https');
const sslAgent = new https.Agent(
  KeepAlive: true,
  rejectUnauthorized: true
);
sslAgent.setMaxListeners(0);
AWS.config.update(
  httpOptions: 
    agent: sslAgent,
  ,
  region: 'us-east-1'
);

module.exports = class S3Zip 
  constructor(params, bucketName = 'default-bucket') 
    this.params = params;
    this.BucketName = bucketName;
  

  async process() 
    const  params, BucketName  = this;
    const s3 = new AWS.S3( apiVersion: '2006-03-01', params:  Bucket: BucketName  );

    // create readstreams for all the output files and store them
    const createReadStream = fs.createReadStream;
    const s3FileDwnldStreams = params.files.map(item => 
      const stream = s3.getObject( Key: item.key ).createReadStream();
      return 
        stream,
        fileName: item.fileName
      
    );

    const streamPassThrough = new Stream.PassThrough();
    // Create a zip archive using streamPassThrough style for the linking request in s3bucket
    const uploadParams = 
      ACL: 'private',
      Body: streamPassThrough,
      ContentType: 'application/zip',
      Key: params.zippedFileKey
    ;

    const s3Upload = s3.upload(uploadParams, (err, data) => 
      if (err) 
        console.error('upload err', err)
       else 
        console.log('upload data', data);
      
    );

    s3Upload.on('httpUploadProgress', progress => 
      // console.log(progress); //  loaded: 4915, total: 192915, part: 1, key: 'foo.jpg' 
    );

    // create the archiver
    const archive = Archiver('zip', 
      zlib:  level: 0 
    );
    archive.on('error', (error) => 
      throw new Error(`$error.name $error.code $error.message $error.path $error.stack`);
    );

    // connect the archiver to upload streamPassThrough and pipe all the download streams to it
    await new Promise((resolve, reject) => 
      console.log("Starting upload of the output Files Zip Archive");

      streamPassThrough.on('close', resolve());
      streamPassThrough.on('end', resolve());
      streamPassThrough.on('error', reject());

      archive.pipe(streamPassThrough);
      s3FileDwnldStreams.forEach((s3FileDwnldStream) => 
        archive.append(s3FileDwnldStream.stream,  name: s3FileDwnldStream.fileName )
      );
      archive.finalize();

    ).catch((error) => 
      throw new Error(`$error.code $error.message $error.data`);
    );

    // Finally wait for the uploader to finish
    await s3Upload.promise();

  

【讨论】：

像魅力一样工作。非常感谢李振琪。我发现这个答案的结构很好，比这里的其他答案更容易理解。

@HardikShah 欢迎您。我很高兴能够提供帮助。

更新：因此这适用于压缩 600-700 个文件（所有文件的大小约为 2mb），但一旦文件数超过该数量，则不会创建 zip，也不会记录错误。所有被压缩的文件都很小~2-3 mb。知道可能出了什么问题吗？

我这边没有问题，即使单个文件达到7MB。而我的一般配置是：内存（3008MB），超时（15分钟）。您可以检查您的常规配置或调试代码。 @HardikShah

@zhenqili，我已经创建了一个 lambda 函数并放置了这个代码和配置，但是 lambda 函数抛出了这个错误“trace”：[“Runtime.ImportModuleError: Error: Cannot find module 'archiver'” could请帮忙解决？

【参考方案4】：

其他解决方案非常适合文件不多（小于~60）。如果他们处理更多文件，他们就会毫无错误地退出。这是因为他们打开了太多的流。

这个解决方案的灵感来自https://gist.github.com/amiantos/16bacc9ed742c91151fcf1a41012445e

这是一个有效的解决方案，即使处理许多文件 (+300) 也能很好地工作，并将预签名的 URL 返回到包含文件的 zip。

主 Lambda：

const AWS = require('aws-sdk');
const S3 = new AWS.S3(
  apiVersion: '2006-03-01',
  signatureVersion: 'v4',
  httpOptions: 
    timeout: 300000 // 5min Should Match Lambda function timeout
  
);
const archiver = require('archiver');
import stream from 'stream';

const UPLOAD_BUCKET_NAME = "my-s3-bucket";
const URL_EXPIRE_TIME = 5*60;

export async function getZipSignedUrl(event) 
  const prefix = `uploads/id123123/`;   //replace this with your S3 prefix
  let files = ["12314123.png", "56787567.png"]  //replace this with your files

  if (files.length == 0) 
    console.log("No files to zip");
    return result(404, "No pictures to download");
  
  console.log("Files to zip: ", files);

  try 
    files = files.map(file => 
        return 
            fileName: file,
            key: prefix + '/' + file,
            type: "file"
        ;
    );
    const destinationKey = prefix + '/' + 'uploads.zip'
    console.log("files: ", files);
    console.log("destinationKey: ", destinationKey);

    await streamToZipInS3(files, destinationKey);
    const presignedUrl = await getSignedUrl(UPLOAD_BUCKET_NAME, destinationKey, URL_EXPIRE_TIME, "uploads.zip");
    console.log("presignedUrl: ", presignedUrl);

    if (!presignedUrl) 
      return result(500, null);
    
    return result(200, presignedUrl);
  
  catch(error) 
    console.error(`Error: $error`);
    return result(500, null);
  

辅助函数：

export function result(code, message) 
  return 
    statusCode: code,
    body: JSON.stringify(
      
        message: message
      
    )
  


export async function streamToZipInS3(files, destinationKey) 
  await new Promise(async (resolve, reject) => 
    var zipStream = streamTo(UPLOAD_BUCKET_NAME, destinationKey, resolve);
    zipStream.on("error", reject);

    var archive = archiver("zip");
    archive.on("error", err => 
      throw new Error(err);
    );
    archive.pipe(zipStream);

    for (const file of files) 
      if (file["type"] == "file") 
        archive.append(getStream(UPLOAD_BUCKET_NAME, file["key"]), 
          name: file["fileName"]
        );
      
    
    archive.finalize();
  )
  .catch(err => 
    console.log(err);
    throw new Error(err);
  );


function streamTo(bucket, key, resolve) 
  var passthrough = new stream.PassThrough();
  S3.upload(
    
      Bucket: bucket,
      Key: key,
      Body: passthrough,
      ContentType: "application/zip",
      ServerSideEncryption: "AES256"
    ,
    (err, data) => 
      if (err) 
        console.error('Error while uploading zip')
        throw new Error(err);
        reject(err)
        return
      
      console.log('Zip uploaded')
      resolve()
    
  ).on("httpUploadProgress", progress => 
    console.log(progress)
  );
  return passthrough;


function getStream(bucket, key) 
  let streamCreated = false;
  const passThroughStream = new stream.PassThrough();

  passThroughStream.on("newListener", event => 
    if (!streamCreated && event == "data") 
      const s3Stream = S3
        .getObject( Bucket: bucket, Key: key )
        .createReadStream();
      s3Stream
        .on("error", err => passThroughStream.emit("error", err))
        .pipe(passThroughStream);

      streamCreated = true;
    
  );

  return passThroughStream;


export async function getSignedUrl(bucket: string, key: string, expires: number, downloadFilename?: string): Promise<string> 
    const exists = await objectExists(bucket, key);
    if (!exists) 
        console.info(`Object $bucket/$key does not exists`);
        return null
    

    let params = 
        Bucket: bucket,
        Key: key,
        Expires: expires,
    ;
    if (downloadFilename) 
        params['ResponseContentDisposition'] = `inline; filename="$encodeURIComponent(downloadFilename)"`; 
    
    
    try 
        const url = s3.getSignedUrl('getObject', params);
        return url;
     catch (err) 
        console.error(`Unable to get URL for $bucket/$key`, err);
        return null;
    
;

【讨论】：

使用这个 sn-p 作为我自己需求的灵感来源。非常有用，效果很好，非常感谢！

干杯！很高兴听到它有效！这就是我把它放在这里的原因！