給定 S3 中的一個(gè)大gzip對(duì)象，python3/boto3 中的內(nèi)存高效（例如流式傳輸）方法是什么來(lái)解壓縮數(shù)據(jù)并將結(jié)果存儲(chǔ)回另一個(gè) S3 對(duì)象？之前也有人問(wèn)過(guò)類(lèi)似的問(wèn)題。然而，所有答案都使用一種方法，其中首先將 gzip 文件的內(nèi)容讀入內(nèi)存（例如ByteIO）。這些解決方案對(duì)于太大而無(wú)法放入主內(nèi)存的對(duì)象來(lái)說(shuō)是不可行的。對(duì)于大型 S3 對(duì)象，需要讀取內(nèi)容，“即時(shí)”解壓縮，然后以某種分塊方式寫(xiě)入不同的 S3 對(duì)象。預(yù)先感謝您的考慮和回復(fù)。
1 回答
躍然一笑
TA貢獻(xiàn)1826條經(jīng)驗(yàn) 獲得超6個(gè)贊
您可以將流方法與boto / s3一起使用，但您必須定義自己的類(lèi)文件對(duì)象 AFAIK。
幸運(yùn)的是，有smart_open可以幫你處理這個(gè)問(wèn)題；它還支持GCS、Azure、HDFS、SFTP等。以下是使用大量銷(xiāo)售數(shù)據(jù)樣本
的 示例：
import boto3
from smart_open import open
session = boto3.Session()? # you need to set auth credentials here if you don't have them set in your environment
chunk_size = 1024 * 1024? # 1 MB
f_in = open("s3://mybucket/2m_sales_records.csv.gz", transport_params=dict(session=session), encoding="utf-8")
f_out = open("s3://mybucket/2m_sales_records.csv", "w", transport_params=dict(session=session))
byte_count = 0
while True:
? ? data = f_in.read(chunk_size)
? ? if not data:
? ? ? ? break
? ? f_out.write(data)
? ? byte_count += len(data)
? ? print(f"wrote {byte_count} bytes so far")
f_in.close()
f_out.close()
示例文件有200 萬(wàn)行，壓縮后為75 MB，未壓縮為238 MB。
我將壓縮文件上傳到mybucket并運(yùn)行下載該文件的代碼，提取內(nèi)存中的內(nèi)容并將未壓縮的數(shù)據(jù)上傳回 S3。
在我的計(jì)算機(jī)上，該過(guò)程大約需要78 秒（高度依賴(lài)于互聯(lián)網(wǎng)連接速度），并且從未使用超過(guò)95 MB的內(nèi)存；我認(rèn)為如果需要的話，您可以通過(guò)覆蓋smart_open中 S3 分段上傳的部分大小來(lái)降低內(nèi)存要求。
DEFAULT_MIN_PART_SIZE = 50 * 1024**2
"""Default minimum part size for S3 multipart uploads"""
MIN_MIN_PART_SIZE = 5 * 1024 ** 2
"""The absolute minimum permitted by Amazon."""
添加回答
舉報(bào)
0/150
提交
取消