python - 如何在 Python 中将 Azure Blob 存储中的 CSV 作为流处理

Question

使用 azure.storage.blob 包获取 StorageStreamDownloader 非常简单:

from azure.storage.blob import BlobServiceClient

blob_service_client = BlobServiceClient.from_connection_string("my azure connection string")
container_client = blob_service_client.get_container_client("my azure container name")
blob_client = container_client.get_blob_client("my azure file name")
storage_stream_downloader = blob_client.download_blob()

处理类似文件的对象很简单，或者更具体地说，我认为，csv包中的字符串返回迭代器(或对象的文件路径):

import csv
from io import StringIO
 
csv_string = """col1, col2
a,b
c,d"""
with StringIO(csv_string) as csv_file:
  for row in csv.reader(csv_file):
    print(row) # or rather whatever I actually want to do on a row by row basis, e.g. ascertain that the file contains a row that meets a certain condition

我正在努力解决的是将流数据从我的 StorageStreamDownloader 获取到 csv.reader() 的方式，以便我可以在每一行到达时对其进行处理而不是等待整个文件下载。

Microsoft docs让我觉得他们的标准有点承保(chunks() 方法没有注释？)，但我看到有一个 readinto() 方法用于读入流。我尝试读入 BytesIO 流，但无法弄清楚如何将数据输出到 csv.reader() 中，而无需将缓冲区输出到新文件并读取该文件文件。这一切都让我觉得这是一件应该可行的事情，但我可能在概念上遗漏了一些明显的东西，也许与 itertools 或 asyncio 有关，或者也许我只是使用适合我需求的 csv 工具有误吗？

Answer 1

基于 Jim Xu 的评论:

stream = blob_client.download_blob()  
with io.BytesIO() as buf:
  stream.readinto(buf)

  # needed to reset the buffer, otherwise, panda won't read from the start
  buf.seek(0)

  data = pd.read_csv(buf)

或

csv_content = blob_client.download_blob().readall()
data = pd.read_csv(io.BytesIO(csv_content ))