有没有办法使用 python sdk 将 azure blob 作为流读取和处理，而不将整个 blob 加载到内存中？

Question

我希望能够使用 python SDK 将 azure blob 像 IO 对象一样处理。据我所知，这需要我：

a) 使用

.readinto()

将 blob 读入 IO 对象（从而将整个未定义大小的 blob 加载到具有 256Mb 内存的容器内的内存中）

b）手动调用

.read()

并设置偏移量和限制（因此需要我准确地知道在给定时间我想读多少内容）

我正在尝试使用

bson.decode_file_iter

读取 gzipped mongodump BSON 文件，所以我不知道在给定的范围内要读取多少字节（至少，不需要自己稍微解码一下 BSON） ，显然，如果我有特别巨大的转储文件（例如，我现在正在使用未压缩的半演出），则解决方案 a) 并不是很好。据我所知，azure blob sdk 中没有任何内容公开这一点。在理想的情况下，我能够通过 gzip 解压缩将 blob 流式传输到 bson 解码函数中。有什么东西可以用来实现这个而无需编写完整的翻译层，或者我应该坚持当前的方法，将 blob 下载到文件，然后使用 gzip.open

 读取下载的文件并将其通过管道传输到

bson.decode_file_iter

？

Answer 1

有没有办法使用 python sdk 将 azure blob 作为流读取和处理，而不将整个 blob 加载到内存中？

您可以使用下面的代码使用 python sdk 将 azure blob 作为流读取和处理，而无需将整个 blob 加载到内存中。

代码：

import io
import gzip
import bson
from azure.storage.blob import BlobServiceClient

class BlobStream(io.RawIOBase):
    def __init__(self, blob_client):
        self.blob_client = blob_client
        self.stream = self.blob_client.download_blob()
        self.stream_iter = self.stream.chunks()
        self.buffer = b""

    def readinto(self, b):
        try:
            chunk = next(self.stream_iter)
            self.buffer += chunk
        except StopIteration:
            pass
        size = len(self.buffer)
        length = min(len(b), size)
        b[:length] = self.buffer[:length]
        self.buffer = self.buffer[length:]
        return length

    def readable(self):
        return True
    
blob_service_client = BlobServiceClient.from_connection_string("DefaultEndpointsProtocol=https;AccountName=venkat123;AccountKey=/Z3A9SDaP7a8DKxolqooke13Z4Uyxxxxx==;EndpointSuffix=core.windows.net")
blob_client = blob_service_client.get_blob_client(container="test", blob="large_sample.gz")

blob_stream = BlobStream(blob_client)

with gzip.GzipFile(fileobj=blob_stream) as gzipped_blob:
    for doc in bson.decode_file_iter(gzipped_blob):
        print(doc)

在此代码中，自定义

BlobStream

 类扩展了 

io.RawIOBase

。该类使用 

download_blob

 对象的 

BlobClient

 方法将 blob 下载到流。

它使用

chunks

 对象的 

StorageStreamDownloader

 方法以块的形式读取流。

readinto

 类的 

BlobStream

 方法将数据从流读取到缓冲区并返回读取的字节数。

通过分块处理流并缓冲数据，代码可以有效地处理大型 blob，而无需立即将整个 blob 加载到内存中。

输出：

{vGAxXP6r4r': '42tt1xFZ4fsHQizfxsEbhIepntrWBcd2KSU7P58RFp0OiQiMZ5kDKIdoKscVazIxXBjtOJAQNv3oLGCGrvLPgiOrDSbjGJ4wXH6h', 'USvIfwuHQ6': 'pOk9xAXWdKuPVcWImLYKRsmDq1fIHCgRfHmUnv5t5tTT3fg6RE0bKmCf6V8UQ9JBYdlxRxMAAxCsdfBeIdNO8PtQOswxewlqnFGZ', 'YNgIcaA3N9': '3A4GLIaLmpjSaRZGecZqUh76xjRGTnBQ4sXB3Bcl7CSh8ycToVWmX81mRlKvVqSnYnDy92fpv7kOD5hxdIG6JCWps5cLm99e3Gqr', 'DLsJ1QOXvJ': 'f6NK561gNY0ZYxJcKO2KoKxbyRjzIgiKG9AjNWes3ti3kSTyuwjKNuhb48MiHMTuGFblLz4ufFpAj76tD550ajTqBbyeODXkyZhc', 'xnip5jrUGV': 'K4OEwJVhkd1gzBwqeq5LDkNNZU2Z2MMr7FSy42SOSn2SBI8xlpvcrbqTZs5JZWBi9abP4ziaE3lqf6FBvyYJ2iQeiYpuWybPRnAE', 'KZbTjdxBcw': 'wYX2ioBqxTbCTzHqmJEOv4fNsDq8wFKAWD8x3I4mX7fmpyI0K9ZODUi0nn7020tgXD021TXuSlhjAyMLaJVy9uhmZ7cu3l4JmyEU', 'POZkbGX1Oo': 'Rstj0ldaOFLy4doyPMXbLR9MH05TRLW820yWWsHAlAzb3uporaLxyON290w6tPLRUiOVJqfr8l1q3HpSc48yxuheZ0cz3pgwRqQl', 'b6zzcCZqvT': 'fISfGw3ftsB6BkmJYosIgKeMsSTHYbWIDrYMfH14LA7hYg4oj852UQ2I76uHC93nltZLIawDKv5GFXZLlcYLQwPmDsuFGC22cpaN', 'fBb6tmprgQ': '8JmO0twpLmNtjNQn7Zn5NeByLzEpSldfPUtCWOQytfcj2aFizu7ma8bWGVRY5Xbg8v5eX8lWJC6k7ddOQRZNoAdSWiCOQBfD8m7J', 'YWd247222J': 'Y05nSM1lcnUrnsmEKndc6hoX6pItvoJFkkE5qYXuSCmgpLnwcfKgY8lUIgXLeG04BUIVqVytNGgpaNHKOyXb7hZcl7uJpgS2tmeG', 'UuA86rFKln': 'nNWuL6SRUoUrkoMCR8dKb5vzFxGi4JVwRbGZHcC6lrCUk5ZRZdkL0VoG1RScRCwE3l1KSWCUaKeWbGbvPpSwFIBJSPvvk1bJZUFy', 'sqH5XlZ6Dv': 'GHtgkrDtxJcQwRY5WSHo1oJwPpjhh8NCkh4E7Wb1qCXUE2SWdaZLpnPxFUx9mJMQgAsf3Rrs2djlvkTL0B3J9xudpCRhv8v9gdM4', 'TSy8HqfeuJ': '9UDVGZQLOhnd7Cgknstfzeth8SM1gGecYDfufj5MWRaWoVGetJ0eVt2unUEgjjVMBumUZgW2jPJFC1h4tPbYlHWs2kvBDRTsHJBZ', 'AeSdgHc3ML': 'ImJquzYHwfV394qlp09i2udfJ0o8NjsXZusyLALjYD3jKMrGvpRmLGpQJQlpKcd5vSXMOKXQFET665ix0wB0vARFodSgddRqDHmS', 'MyWBr0uG5W': 'lv00rWQ42e4yrWen8V9VL9KgQPYCC9EGh5kifN52rjzgdSO1iRtif7b32qZ5weSaQz6oHfHQflUhLPSMr6ZpwpI00kylzt6RK7WA', 'XLGO6tFBwV': 'aiWVzwNAQt4QoRfJBUWNPykxAPlWsY6vQFDg5UhafPybRTE970fmDKYnNRRuJpvz3xntNpargubjM3mYDqX6vBtRs37HeIWY5soX'}

参考： azure.storage.blob.StorageStreamDownloader 类 |微软学习