You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何使用Python的put_block分块上传文件至Azure Blob?

解决Azure Blob分块上传(put_block)的问题

我明白你想用分块上传来避免大文件失败的困扰,之前的代码确实存在几个关键问题导致无法正常工作,比如旧SDK的使用、block_id的编码格式错误,还有缺少必要的初始化步骤。下面给你一个基于最新Azure Blob存储SDK v12的可行方案,这是微软当前推荐的版本:

先分析你之前代码的问题

  • 第一段代码用的是create_blob_from_path,这是直接整文件上传,没有用到分块逻辑,不符合你的需求;
  • 第二段代码使用了已被弃用的旧版BlockBlobService,而且block_id的处理有误:Azure要求block_id必须是Base64编码的字符串,你直接编码数字bytes的方式可能导致格式不兼容,另外收集block_ids的格式也不对,新版SDK需要用BlobBlock对象列表。

完整可行的分块上传代码

import base64
from azure.storage.blob import BlobServiceClient, BlobBlock

def upload_large_file_in_chunks():
    # 替换为你的Azure存储账户信息
    account_name = "your-account-name"
    account_key = "your-account-key"
    container_name = "your-container-name"
    blob_name = "image1.jpg"  # 上传后的Blob名称
    file_path = "./images/python.jpg"  # 本地文件路径
    chunk_size = 4 * 1024 * 1024  # 推荐每个块4MB(Azure要求最小1MB,最大100MB)

    # 初始化Blob服务客户端(新版SDK)
    connect_str = f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)

    # 获取容器客户端(如果容器不存在则创建)
    container_client = blob_service_client.get_container_client(container_name)
    if not container_client.exists():
        container_client.create_container()

    # 获取Blob客户端
    blob_client = container_client.get_blob_client(blob_name)

    block_list = []
    index = 0

    with open(file_path, "rb") as f:
        while True:
            chunk_data = f.read(chunk_size)
            if not chunk_data:
                break  # 读取完毕,退出循环

            # 生成符合要求的block_id:Base64编码的字符串,建议用带序号的标识保证唯一性
            block_id = base64.b64encode(f"block-{index:06d}".encode()).decode("utf-8")
            # 上传单个块
            blob_client.stage_block(block_id=block_id, data=chunk_data)
            # 将块ID添加到列表中(需要用BlobBlock对象)
            block_list.append(BlobBlock(block_id=block_id))
            index += 1
            print(f"已上传第 {index} 个块")

    # 合并所有块,完成文件上传
    blob_client.commit_block_list(block_list)
    print(f"文件 {file_path} 已成功分块上传至Blob {blob_name}")

# 调用函数
upload_large_file_in_chunks()

关键注意事项

  • SDK版本:确保你安装的是新版SDK,执行pip install azure-storage-blob即可,旧版的azure-storage包已经被弃用;
  • 块大小:每个块的大小建议设置为4MB到100MB之间,太小会增加API调用次数,太大则失去分块的意义;
  • Block ID要求:必须是Base64编码的字符串,长度不能超过64字节,用带序号的命名方式可以避免重复;
  • 异常处理:如果需要更健壮的代码,可以添加try-except块来处理网络异常,比如上传失败时重试单个块。

内容的提问来源于stack exchange,提问作者Shakir Shakeel

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.13 08:47:23