如何通过Python获取所有Azure Blob内容并存入DataFrame

问题场景 #

需要列出Azure Blob Storage指定容器内的所有Blob，将其中所有CSV格式文件的内容读取后存入Pandas DataFrame。当前已选用操作最简便的BlobServiceClient实现Blob列表拉取，基础代码如下：

#!/usr/bin/env python3

import os
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
from pathlib import Path
from io import StringIO
import pandas as pd

def main():
    connect_str = os.environ['AZURE_CONNECT_STR']
    container   = os.environ['CONTAINER']

    print(connect_str + "\n")
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)
    container_client = blob_service_client.get_container_client(container)
    blob_list = container_client.list_blobs()
    for blob in blob_list:
        print("\t" + blob.name)

if __name__ == "__main__":
    main()

最新版Azure Blob Storage Python客户端没有直接在列表遍历阶段返回Blob内容的方法，仅需补充少量代码即可完成内容拉取，不需要更换其他操作更繁琐的SDK客户端。

补充实现代码 #

在遍历Blob的循环内，通过容器客户端获取单个Blob的操作客户端，调用下载方法拉取内容后转成DataFrame即可，修改后的完整可运行代码如下：

#!/usr/bin/env python3

import os
from azure.storage.blob import BlobServiceClient
from io import StringIO
import pandas as pd

def main():
    connect_str = os.environ['AZURE_CONNECT_STR']
    container_name = os.environ['CONTAINER']

    blob_service_client = BlobServiceClient.from_connection_string(connect_str)
    container_client = blob_service_client.get_container_client(container_name)
    
    # 用字典存储所有CSV对应的DataFrame，key为Blob文件名
    blob_df_map = {}
    blob_list = container_client.list_blobs()

    for blob in blob_list:
        # 跳过非CSV文件
        if not blob.name.lower().endswith('.csv'):
            continue
        print(f"正在读取Blob: {blob.name}")
        # 获取单个Blob的客户端
        blob_client = container_client.get_blob_client(blob=blob.name)
        # 下载Blob内容，解码为字符串
        blob_content = blob_client.download_blob().readall().decode('utf-8')
        # 转成DataFrame存入字典
        blob_df_map[blob.name] = pd.read_csv(StringIO(blob_content))

    # 后续可以直接通过blob_df_map[blob名称]取对应DataFrame
    # 示例：打印第一个读取到的CSV的前5行
    first_blob_name = next(iter(blob_df_map))
    print(f"{first_blob_name} 数据预览：")
    print(blob_df_map[first_blob_name].head())

if __name__ == "__main__":
    main()

关键逻辑说明 #

• 全程基于已在使用的BlobServiceClient体系实现，没有额外的鉴权或客户端初始化开销
• download_blob()是新版SDK中拉取Blob内容的标准方法，返回的流式对象调用readall()可以一次性拉取全量内容，针对常规大小的CSV文件足够使用
• 增加了.csv后缀判断，避免误读取容器内其他格式的文件导致解析报错
• 如果CSV文件使用GBK等其他编码，把decode('utf-8')里的编码参数改成对应值即可
• 如果单个CSV文件体积超过1G，建议将readall()替换为分块读取逻辑，避免内存占用过高

内容的提问来源于stack exchange，提问作者ChrisAdkin