Потоковая передача огромного файла gzip из s3 с использованием boto3 python [duplicate]

Я полагаю, что, по крайней мере, некоторые из людей, которые видят этот вопрос, будут похожи на меня, и мне захочется создать поток из boto по строке (или запятую через запятую или любой другой разделитель). Вот простой способ сделать это:

def getS3ResultsAsIterator(self, aws_access_info, key, prefix):        
    s3_conn = S3Connection(**aws_access)
    bucket_obj = s3_conn.get_bucket(key)
    # go through the list of files in the key
    for f in bucket_obj.list(prefix=prefix):
        unfinished_line = ''
        for byte in f:
            byte = unfinished_line + byte
            #split on whatever, or use a regex with re.split()
            lines = byte.split('\n')
            unfinished_line = lines.pop()
            for line in lines:
                yield line

Ответ @ garnaat выше по-прежнему велик и на 100% прав. Надеюсь, мой все еще помогает кому-то.

Это мое решение обтекающего потокового тела:

import io
class S3ObjectInterator(io.RawIOBase):
    def __init__(self, bucket, key):
        """Initialize with S3 bucket and key names"""
        self.s3c = boto3.client('s3')
        self.obj_stream = self.s3c.get_object(Bucket=bucket, Key=key)['Body']

    def read(self, n=-1):
        """Read from the stream"""
        return self.obj_stream.read() if n == -1 else self.obj_stream.read(n)

Пример использования:

obj_stream = S3ObjectInterator(bucket, key)
for line in obj_stream:
    print line

Другие ответы в этой теме связаны с boto, но S3.Object больше не повторяется в boto3. Таким образом, следующее НЕ РАБОТАЕТ, оно выдает сообщение об ошибке TypeError: 's3.Object' object is not iterable:

    s3 = boto3.session.Session(profile_name=my_profile).resource('s3')
    s3_obj = s3.Object(bucket_name=my_bucket, key=my_key)

    with io.FileIO('sample.txt', 'w') as file:
        for i in s3_obj:
            file.write(i)

В boto3 содержимое объекта доступно в S3.Object.get()['Body'], которое также не является итерируемым, поэтому следующий еще НЕ РАБОТАЕТ:

    body = s3_obj['Body']
    with io.FileIO('sample.txt', 'w') as file:
        for i in body:
            file.write(i)

Итак, альтернативой является использование метода чтения, но это загружает объект WHOLE S3 в память, который при работе с большими файлами не всегда возможен:

    body = s3_obj['Body']
    with io.FileIO('sample.txt', 'w') as file:
        for i in body.read():
            file.write(i)

Но метод read позволяет передать в amt параметр, определяющий количество байтов, которые мы хотим прочитать из базового потока. Этот метод можно многократно вызывать до тех пор, пока весь поток не будет прочитан:

    body = s3_obj['Body']
    with io.FileIO('sample.txt', 'w') as file:
        while file.write(body.read(amt=512)):
            pass

Копаем в код botocore.response.StreamingBody, понимаем, что базовый поток также доступен, поэтому мы могли бы выполнять итерацию следующим образом:

    body = s3_obj['Body']
    with io.FileIO('sample.txt', 'w') as file:
        for b in body._raw_stream:
            file.write(b)

В то время как googling я также видел некоторые ссылки, которые могут быть использованы, но я не пробовал:

• WrappedStreamingBody
• Другой связанный поток
• Проблема в boto3 github для запроса StreamingBody - это правильный поток - который был закрыт !!!