在Python中从AWS S3读取gzip文件的内容

Question

我正在尝试从我在AWS中运行的Hadoop进程中读取一些日志.日志存储在S3文件夹中,并具有以下路径.

bucketname = name key = y/z/stderr.gz这里Y是集群ID,z是文件夹名称.这两者都充当AWS中的文件夹(对象).所以完整路径就像x/y/z/stderr.gz.

现在我想解压缩.gz文件并读取文件的内容.我不想将此文件下载到我的系统想要将内容保存在python变量中.

这是我到现在为止所尝试的.

bucket_name = "name"
key = "y/z/stderr.gz"
obj = s3.Object(bucket_name,key)
n = obj.get()['Body'].read()

这给了我一种不可读的格式.我也试过了

n = obj.get()['Body'].read().decode('utf-8')

这给出了错误utf8'编解码器无法解码位置1中的字节0x8b:无效的起始字节.

我也试过了

gzip = StringIO(obj)
gzipfile = gzip.GzipFile(fileobj=gzip)
content = gzipfile.read()

这将返回错误IOError:不是gzip压缩文件

不确定如何解码此.gz文件.

编辑 - 找到解决方案.需要传递n并使用BytesIO

gzip = BytesIO(n)

Answer 1

这是旧的，但您不再需要中间的 BytesIO 对象（至少在我boto3==1.9.223和python3.7）

import boto3
import gzip

s3 = boto3.resource("s3")
obj = s3.Object("YOUR_BUCKET_NAME", "path/to/your_key.gz")
with gzip.GzipFile(fileobj=obj.get()["Body"]) as gzipfile:
    content = gzipfile.read()
print(content)

Answer 2

您可以使用 AWS S3 SELECT Object Content读取 gzip 内容

S3 Select 是一项 Amazon S3 功能，旨在仅从对象中提取您需要的数据，这可以显着提高性能并降低需要访问 S3 中数据的应用程序的成本。

Amazon S3 Select 适用于以 Apache Parquet 格式存储的对象、JSON 数组以及用于 CSV 和 JSON 对象的 BZIP2 压缩。

参考：https : //docs.aws.amazon.com/AmazonS3/latest/dev/selecting-content-from-objects.html

from io import StringIO
import boto3
import pandas as pd

bucket = 'my-bucket'
prefix = 'my-prefix'

client = boto3.client('s3')

for object in client.list_objects_v2(Bucket=bucket, Prefix=prefix)['Contents']:
    if object['Size'] <= 0:
        continue

    print(object['Key'])
    r = client.select_object_content(
            Bucket=bucket,
            Key=object['Key'],
            ExpressionType='SQL',
            Expression="select * from s3object",
            InputSerialization = {'CompressionType': 'GZIP', 'JSON': {'Type': 'DOCUMENT'}},
            OutputSerialization = {'CSV': {'QuoteFields': 'ASNEEDED', 'RecordDelimiter': '\n', 'FieldDelimiter': ',', 'QuoteCharacter': '"', 'QuoteEscapeCharacter': '"'}},
        )

    for event in r['Payload']:
        if 'Records' in event:
            records = event['Records']['Payload'].decode('utf-8')
            payloads = (''.join(r for r in records))
            try:
                select_df = pd.read_csv(StringIO(payloads), error_bad_lines=False)
                for row in select_df.iterrows():
                    print(row)
            except Exception as e:
                print(e)

Answer 3

@Amit，我试图做同样的事情来测试解码文件，并使您的代码经过一些修改即可运行。我只需要删除函数def，返回值并重命名gzip变量，因为该名称正在使用中。

import json
import boto3
from io import BytesIO
import gzip

try:
     s3 = boto3.resource('s3')
     key='YOUR_FILE_NAME.gz'
     obj = s3.Object('YOUR_BUCKET_NAME',key)
     n = obj.get()['Body'].read()
     gzipfile = BytesIO(n)
     gzipfile = gzip.GzipFile(fileobj=gzipfile)
     content = gzipfile.read()
     print(content)
except Exception as e:
    print(e)
    raise e

Answer 4

就像我们对变量所做的那样，当我们使用 io module\xe2\x80\x99s Byte IO 操作时，数据可以以字节形式保存在内存缓冲区中。

\n\n

下面是一个示例程序来演示这一点：

\n\n

mport io\n\nstream_str = io.BytesIO(b"JournalDev Python: \\x00\\x01")\nprint(stream_str.getvalue())\n

\n\n

该getvalue()函数将缓冲区中的值作为字符串获取。

\n\n

所以，@Jean-Fran\xc3\xa7oisFabre 的答案是正确的，你应该使用

\n\n

gzip = BytesIO(n)\n

\n\n

有关更多信息，请阅读以下文档：

\n\n

https://docs.python.org/3/library/io.html

\n