Windows环境下Python下载Azure Blob Storage子目录并镜像结构的最优方案

核心问题解决思路 #

针对你遇到的官方API用法困惑、旧方案过时、本地目录重建难题，直接给你当前基于最新Azure Blob SDK的靠谱实现，完全适配Windows环境。

优化后的完整代码 #

import pathlib
from azure.storage.blob import BlobServiceClient

# 替换为你的存储账户连接字符串
conn_str = "your_storage_connection_string"
# 初始化容器客户端
container_client = BlobServiceClient.from_connection_string(conn_str).get_container_client("container_name")

# 云端目标子目录前缀（确保结尾带/，精准匹配嵌套目录下的所有内容）
target_blob_prefix = "dir_1/sub_dir_2/desired_dir/"
# 本地存储根目录
local_root_dir = pathlib.Path("local_directory/")

# 遍历目标目录下的所有Blob
for blob in container_client.list_blobs(name_starts_with=target_blob_prefix):
    # 跳过云端的虚拟目录占位符（以/结尾的空Blob）
    if blob.name.endswith("/"):
        continue
    
    # 计算Blob相对于目标目录的路径，用于镜像本地结构
    relative_blob_path = pathlib.Path(blob.name[len(target_blob_prefix):])
    # 拼接本地完整文件路径
    local_file_path = local_root_dir / relative_blob_path
    
    # 自动创建多层嵌套父目录，已存在则忽略
    local_file_path.parent.mkdir(parents=True, exist_ok=True)
    
    # 获取Blob客户端，调用官方下载方法
    blob_client = container_client.get_blob_client(blob)
    # 官方download_to_file方法无需手动处理流，自动适配Windows路径
    blob_client.download_to_file(local_file_path)
    print(f"已完成下载：{blob.name} -> {local_file_path}")

关键细节说明 #

1. 官方download_to_file的self参数：这个是类方法的实例绑定参数，你不用手动传。当你通过container_client.get_blob_client(blob)拿到BlobClient实例后，直接调用blob_client.download_to_file(本地路径)就行，self会自动关联到当前实例。
2. 虚拟目录处理：Azure Blob没有真实文件夹，那些看起来像目录的路径其实是带/结尾的空Blob占位符，代码里判断跳过这些，避免本地生成空文件。
3. Parquet/Snappy结构适配：这类文件在云端是按xxx.parquet/part-xxxx.snappy的路径存储的，代码会自动在本地重建对应的子目录，把每个分片文件放到正确位置，完全镜像云端结构。
4. Windows路径兼容：用pathlib处理路径会自动转换为Windows的\分隔符，不用手动替换，避免路径错误。

额外实用优化 #

• 并发下载：如果Blob数量多，用多线程能大幅提升速度，示例代码如下：

  from concurrent.futures import ThreadPoolExecutor
  
  def download_blob(blob):
      if blob.name.endswith("/"):
          return
      relative_blob_path = pathlib.Path(blob.name[len(target_blob_prefix):])
      local_file_path = local_root_dir / relative_blob_path
      local_file_path.parent.mkdir(parents=True, exist_ok=True)
      blob_client = container_client.get_blob_client(blob)
      blob_client.download_to_file(local_file_path)
      print(f"已完成下载：{blob.name}")
  
  # 开启10个线程并发下载
  with ThreadPoolExecutor(max_workers=10) as executor:
      executor.map(download_blob, container_client.list_blobs(name_starts_with=target_blob_prefix))
  

• 断点续传：针对大文件，可在download_to_file中设置max_concurrency参数（如max_concurrency=4），或者捕获异常后记录未完成的Blob，后续重试。
• 日志替代打印：用Python的logging模块替换print，方便后续排查问题。

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