AWS Lambda：如何在 S3 存储桶中提取 tgz 文件并将其放入另一个 S3 存储桶中

Question

我有一个名为“Source”的 S3 存储桶。许多“.tgz”文件被实时推送到该存储桶中。我编写了一个 Java 代码来提取“.tgz”文件并将其推送到“目标”存储桶中。我将我的代码推送为 Lambda 函数。我在我的 Java 代码中获得了作为 InputStream 的“.tgz”文件。如何在 Lambda 中提取它？我无法在 Lambda 中创建文件，它会在 JAVA 中抛出“FileNotFound(Permission Denied)”。

AmazonS3 s3Client = new AmazonS3Client();
S3Object s3Object = s3Client.getObject(new GetObjectRequest(srcBucket, srcKey));
InputStream objectData = s3Object.getObjectContent();
File file = new File(s3Object.getKey());
OutputStream writer = new BufferedOutputStream(new FileOutputStream(file)); <--- It throws FileNotFound(Permission denied) here

Answer 1

由于其中一个响应是用 Python 编写的，因此我提供了这种语言的替代解决方案。

使用 /tmp 文件系统的解决方案的问题是，AWS 只允许在其中存储 512 MB (read more)。为了解压或解压较大的文件，最好使用 io 包和BytesIO 类并纯粹在内存中处理文件内容。 AWS 允许为 Lambda 分配高达 3GB 的 RAM，这显着扩展了最大文件大小。我成功测试了 1GB S3 文件的解压。

在我的例子中，将 ~2000 个文件从 1GB tar 文件解压缩到另一个 S3 存储桶需要 140 秒。它可以通过利用多个线程将未解压缩的文件上传到目标 S3 存储桶来进一步优化。

下面的示例代码展示了单线程解决方案：

import boto3
import botocore
import tarfile

from io import BytesIO
s3_client = boto3.client('s3')

def untar_s3_file(event, context):

    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    input_tar_file = s3_client.get_object(Bucket = bucket, Key = key)
    input_tar_content = input_tar_file['Body'].read()

    with tarfile.open(fileobj = BytesIO(input_tar_content)) as tar:
        for tar_resource in tar:
            if (tar_resource.isfile()):
                inner_file_bytes = tar.extractfile(tar_resource).read()
                s3_client.upload_fileobj(BytesIO(inner_file_bytes), Bucket = bucket, Key = tar_resource.name)

Answer 2

import boto3
import tarfile
from tarfile import TarInfo
import tempfile

s3_client = boto3.client('s3')
s3_resource=boto3.resource('s3')
def lambda_handler(event, context):
    bucket =event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    new_bucket='uncompressed-data' #new bucket name
    new_key=key[:-4]
    try:
        with tempfile.SpooledTemporaryFile(mode='w+t') as temp:
            s3_client.download_fileobj(bucket,key, temp)
            temp.seek(0)
            tar=tarfile.open(mode="r:gz", fileobj = temp)
            for TarInfo in tar:
                file_save=tar.extractfile(TarInfo.name)
                s3_client.upload_fileobj(file_save,new_bucket,new_key)
            tar.close()
            temp.close()
    except Exception as e:
        print(e)
        raise e

使用 Python 3.6 并触发带有后缀“.tgz”的 obejctcreated(all) 事件。希望对您有所帮助。

Answer 3

不要使用File 或FileOutputStream，使用s3Client.putObject()。要读取 tgz 文件，您可以使用 Apache Commons Compress。示例：

ArchiveInputStream tar = new ArchiveInputStreamFactory().
    createArchiveInputStream("tar", new GZIPInputStream(objectData));
ArchiveEntry entry;
while ((entry = tar.getNextEntry()) != null) {
    if (!entry.isDirectory()) {
        byte[] objectBytes = new byte[entry.getSize()];
        tar.read(objectBytes);
        ObjectMetadata metadata = new ObjectMetadata();
        metadata.setContentLength(objectBytes.length);
        metadata.setContentType("application/octet-stream");
        s3Client.putObject(destBucket, entry.getName(), 
            new ByteArrayInputStream(objectBytes), metadata);
    }
}