来自 io.BytesIO 流的 numpy.load

Question

我在 Azure Blob 存储中保存了 numpy 数组，我正在将它们加载到这样的流中：stream = io.BytesIO() store.get_blob_to_stream(container, 'cat.npy', stream)我从中知道stream.getvalue()流包含用于重建数组的元数据。这是前 150 个字节：b"\x93NUMPY\x01\x00v\x00{'descr': '|u1', 'fortran_order': False, 'shape': (720, 1280, 3), }                                                  \n\xc1\xb0\x94\xc2\xb1\x95\xc3\xb2\x96\xc4\xb3\x97\xc5\xb4\x98\xc6\xb5\x99\xc7\xb6\x9a\xc7"是否可以使用numpy.load或通过其他一些简单的方法加载字节流？我可以将阵列保存到磁盘并从磁盘加载它，但出于多种原因我想避免这种情况......编辑：只是为了强调，输出需要是一个 numpy 数组，其形状和数据类型在流的第 128 个字节中指定。

Answer 1

我尝试使用几种方法来实现您的需求。

这是我的示例代码。

from azure.storage.blob.baseblobservice import BaseBlobService

import numpy as np

account_name = '<your account name>'

account_key = '<your account key>'

container_name = '<your container name>'

blob_name = '<your blob name>'

blob_service = BaseBlobService(

    account_name=account_name,

    account_key=account_key

)

示例 1. 使用 sas 令牌生成 blob url 以通过以下方式获取内容 requests

from azure.storage.blob import BlobPermissions

from datetime import datetime, timedelta

import requests

sas_token = blob_service.generate_blob_shared_access_signature(container_name, blob_name, permission=BlobPermissions.READ, expiry=datetime.utcnow() + timedelta(hours=1))

print(sas_token)

url_with_sas = blob_service.make_blob_url(container_name, blob_name, sas_token=sas_token)

print(url_with_sas)

r = requests.get(url_with_sas)

dat = np.frombuffer(r.content)

print('from requests', dat)

示例 2. 通过以下方式将 blob 的内容下载到内存中 BytesIO

import io

stream = io.BytesIO()

blob_service.get_blob_to_stream(container_name, blob_name, stream)

dat = np.frombuffer(stream.getbuffer())

print('from BytesIO', dat)

示例 3. 使用numpy.fromfilewithDataSource打开带有 sas 令牌的 blob url，它实际上会将 blob 文件下载到本地文件系统中。

ds = np.DataSource()

# ds = np.DataSource(None)  # use with temporary file

# ds = np.DataSource(path) # use with path like `data/`

f = ds.open(url_with_sas)

dat = np.fromfile(f)

print('from DataSource', dat)

我认为示例 1 和 2 更适合您。

Answer 2

当涉及到 np.savez 时，上述解决方案通常需要工作。

上传到存储：

import io    

import numpy as np    

stream = io.BytesIO()  

arr1 = np.random.rand(20,4)  

arr2 = np.random.rand(20,4)  

np.savez(stream, A=arr1, B=arr2)  

block_blob_service.create_blob_from_bytes(container, 

                                          "my/path.npz", 

                                          stream.getvalue())

从存储下载：

from numpy.lib.npyio import NpzFile 

stream = io.BytesIO()  

block_blob_service.get_blob_to_stream(container, "my/path.npz", stream)  

ret = NpzFile(stream, own_fid=True, allow_pickle=True)  

print(ret.files)  

""" ['A', 'B'] """  

print(ret['A'].shape)  

""" (20, 4) """  

Answer 3

有点晚了，但如果有人想使用 numpy.load 执行此操作，这里是代码（Azure SDK v12.8.1）：

from azure.storage.blob import BlobServiceClient

import io

import numpy as np

# define your connection parameters

connect_str = ''

container_name = ''

blob_name = ''

blob_service_client = BlobServiceClient.from_connection_string(connect_str)

blob_client = blob_service_client.get_blob_client(container=container_name,

                                                  blob=blob_name)

# Get StorageStreamDownloader

blob_stream = blob_client.download_blob()

stream = io.BytesIO()

blob_stream.download_to_stream(stream)

stream.seek(0)

# Load form io.BytesIO object

data = np.load(stream, allow_pickle=False)

print(data.shape)