python - 将流数据从 Azure Blob 存储读取到 Databricks 中

Question

我正在尝试从 databricks 中的 blob 存储读取文件，通过数据帧进行一些计算并将数据帧写入 cassandra。目前文件是 append blob 类型存储，就像文档页面 https://learn.microsoft.com/en-us/azure/databricks/kb/data-sources/wasb-check-blob-types 中所述。 ，我无法将它们挂载到 DBFS 中。

在本地计算机上工作时，我可以通过适用于 Python 的 Azure 存储 SDK 下载它，但如果我尝试在 databricks 上执行相同的操作，则会收到以下错误:

Unable to stream download: ("Connection broken: ConnectionResetError(104, 'Connection reset by peer')", ConnectionResetError(104, 'Connection reset by peer'))

我正在运行的代码如下

file_dir = data.strftime("%Y%m")
file_name = data.strftime("%Y%m%d") + ".csv"

local_path = f"dbfs:/FileStore/{file_dir}"
local_file = f"{local_path}/{file_name}"
blob_name = f"{file_dir}/{file_name}"

blob_service_client_instance = BlobServiceClient(account_url="https://"+STORAGEACCOUNTURL+".blob.core.windows.net", credential=STORAGEACCOUNTKEY)
blob_client_instance = blob_service_client_instance.get_blob_client(CONTAINERNAME, blob_name, snapshot=None)

with open(local_file, "wb") as my_blob:
           blob_data = blob_client_instance.download_blob(max_concurrency=10)
           blob_data.readinto(my_blob)

另一方面，如果我尝试不将其写入 dbfs:/FileStore/{file_dir} 中，而是使用 os.getcwd() 写入数据 block 的工作类别中我能够写入它，但无法读取该文件(并且从这里读取可能是一个很好的解决方法)。

Answer 1

更改了方法并将文件直接流式传输到数据帧中。答案是由于这篇文章Read csv from Azure blob Storage and store in a DataFrame ，来自“lsl__”的答案。

from io import BytesIO
import pandas as pd

blob_service_client = BlobServiceClient(account_url="https://"+STORAGEACCOUNTURL+".blob.core.windows.net", credential=STORAGEACCOUNTKEY)
container_client = blob_service_client.get_container_client(CONTAINERNAME)
blob_client = container_client.get_blob_client(blob_name)

                
with BytesIO() as input_blob:
  blob_client.download_blob().download_to_stream(input_blob)
  input_blob.seek(0)
  df = pd.read_csv(input_blob)