时间戳

首页 > 解决方案 > 如何使用 StorageStreamDownloader 从 blob 流式下载并将流式上传到不同的 blob？

python - 如何使用 StorageStreamDownloader 从 blob 流式下载并将流式上传到不同的 blob？

问题描述

我相信我有一个非常简单的要求，解决方案让我感到困惑。我是azure-python-sdk的新手，并且在其新的 blob 流功能方面收效甚微。

一些上下文

我已经使用 Java SDK 好几年了。每个CloudBlockBlob对象都有一个BlobInputStream和一个BlobOutputStream对象。当BlobInputStream打开 a 时，可以调用它的许多函数（最显着的是它的read()函数）以真正的流式方式检索数据。A BlobOutputStream，一旦被检索到，就有一个write(byte[] data)函数，在该函数中，人们可以根据需要不断地写入数据，直到调用close()函数。所以，我很容易做到：

  1. 获取一个CloudBlockBlob对象，打开它BlobInputStream并从本质上取回一个InputStream“绑定”到CloudBlockBlob. 它通常维护 4MB 的数据——至少，我是这么理解的。当从其缓冲区中读取一定数量的数据时，会引入新的（相同数量的）数据，因此它始终具有大约 4MB 的新数据（直到检索到所有数据）。
  2. 对该数据执行一些操作。
  3. 检索CloudBlockBlob我要上传到的对象，获取它BlobOutputStream，然后将我对其进行了一些操作的数据写入其中。

一个很好的例子是如果我想压缩一个文件。我有一个GzipStreamReader可以接受 anBlobInputStream和 an的课程BlobOutputStream。BlobInputStream它会从BlobOutputStream. 它可以根据需要多次调用 write()；当它完成一整天的阅读时，它会关闭输入和输出流，一切都很好。

现在为 Python

现在，Python SDK 有点不同，显然是有充分理由的；该io模块的工作方式与 JavaInputStream和OutputStream类（Blob{Input/Output}Stream类继承自它们）不同。我一直在努力理解流在 Azure 的 python SDK 中的真正工作原理。首先，我只是想看看StorageStreamDownloader类是如何工作的。看起来像是我正在从中读取数据StorageStreamDownloader的对象的“连接” 。如果我想将数据放入流中，我会创建一个新的并将该流传递给 readinto方法。BlockBlobio.BytesIO()StorageStreamDownloader

对于上传，我会调用BlobClient的上传方法。upload 方法接受data类型为 的参数Union[Iterable[AnyStr], IO[AnyStr]]。

我不想对我所了解的内容进行过多的详细介绍，因为我所了解的以及我所做的一切都让我一无所获。我怀疑我期待的东西只有 Java SDK 提供。但是，总的来说，这是我遇到的问题：

  1. 当我调用download_blob时，我会返回StorageStreamDownloaderblob 中的所有数据。一些调查表明，我可以使用offset和length来下载我想要的数据量。也许我可以用 a 调用一次download_blob(offset=0, length=4MB)，处理我返回的数据，然后再次调用download_bloc(offset=4MB, length=4MB)，处理数据等。这是不利的。我可以做的另一件事是利用max_chunk_get_size参数BlobClient并打开validate_content标志（使其为真），以便StorageStreamDownloader仅下载 4mb。但这一切都导致了几个问题：这并不是真正从stream对象流式传输的。我还是得打download几次readinto电话。好吧，如果不是第二个问题，我会这样做：
  2. 我如何流式传输上传？上传可以采用流。但是，如果流本身不自动更新，那么我只能上传一次，因为我处理的所有 blob 都必须是BlockBlobs. 该upload_function函数的文档说我可以提供一个参数overwrite：

关键字 bool overwrite：要上传的 blob 是否应该覆盖当前数据。如果为 True，upload_blob 将覆盖现有数据。如果设置为 False，则操作将失败并出现 ResourceExistsError。上面的例外是附加 blob 类型：如果设置为 False 并且数据已经存在，则不会引发错误并且数据将附加到现有 blob。如果设置 overwrite=True，则现有的附加 blob 将被删除，并创建一个新的。默认为假。

  1. 这是有道理BlockBlobs的，因为一旦写入，就无法再次写入。因此，AFAIK，您不能“流式传输”上传。如果我不能有一个直接绑定到 blob 或保存所有数据的流对象，那么 upload() 函数将在完成后立即终止，对吗？

好的。我确定我错过了一些重要的东西。io对于Python中的模块，我也有些无知。尽管我已经使用 Python 进行了很长时间的开发，但我从来没有真正需要过于密切地处理那个模块。我确信我遗漏了一些东西，因为这个功能非常基本，并且存在于我所知道的所有其他 azure SDK 中。

回顾一下

老实说，我上面所说的一切都可以忽略，只阅读这部分；我只是想表明我已经做了一些尽职调查。我想知道如何从 blob 流式传输数据，处理我在流中获取的数据，然后上传该数据。我不能一次接收一个blob中的所有数据。Blob 可能超过 1GB 和所有漂亮的东西。老实说，我会喜欢一些示例代码，这些示例代码显示：

  1. 从流中的 blob 中检索一些数据（一次调用中接收的数据不应超过 10MB）。
  2. 压缩该流中的数据。
  3. 将数据上传到 blob。

这应该适用于所有大小的 blob；它的 1MB 或 10MB 或 10GB 都无关紧要。第 2 步可以是任何东西；它也可以什么都不是。只要数据被下载，插入到流中，然后上传，那就太好了。当然，另一个极其重要的限制是每次“下载”的数据量不应超过 10MB。

我希望这是有道理的！我只想流式传输数据。这不应该那么难。

编辑：

有些人可能想关闭它并声称该问题是重复的。我忘记包含一些非常重要的内容：我目前使用的是最新的、mot 最新的 azure-sdk 版本。我azure-storage-blob的包的版本是12.5.0. 还有其他问题类似于我对严重过时的版本提出的问题。我已经搜索了其他答案，但没有找到任何12+版本。

标签： pythonstreamazure-blob-storage

解决方案

如果要分块下载azure blob，处理每个chunk数据并将每个chunk数据上传到azure blob，请参考以下代码

import io
import os
from azure.storage.blob import BlobClient, BlobBlock
import uuid
key = '<account key>'

source_blob_client = BlobClient(account_url='https://andyprivate.blob.core.windows.net',
                                container_name='',
                                blob_name='',
                                credential=key,
                                max_chunk_get_size=4*1024*1024, # the size of chunk is 4M
                                max_single_get_size=4*1024*1024)

des_blob_client = BlobClient(account_url='https://<account name>.blob.core.windows.net',
                             container_name='',
                             blob_name='',
                             credential=key)
stream = source_blob_client.download_blob()
block_list = []
#read data in chunk
for chunk in stream.chunks():
    #process your data 
    
    # use the put block rest api to upload the chunk to azure storage
    blk_id = str(uuid.uuid4())
    des_blob_client.stage_block(block_id=blk_id, data=<the data after you process>)
    block_list.append(BlobBlock(block_id=blk_id))

#use the put blobk list rest api to ulpoad the whole chunk to azure storage and make up one blob
des_blob_client.commit_block_list(block_list)

另外，如果你只是想将一个blob从存储位置复制到另一个存储位置，你可以直接使用该方法start_copy_from_url 在此处输入图像描述

推荐阅读