Azure Table Storage备份方案、AzCopy权限问题及Python实现咨询

Azure 表存储目前未提供与 Blob 存储对齐的原生即时点备份、软删除能力，通过自定义脚本完成定期备份是当前生产环境的主流实践，完全可以实现遍历指定范围的表、导出到目标Blob容器的需求。

前置条件配置 #

这部分配置同时可以解决之前使用azCopy时遇到的权限报错问题：

• 依赖安装：安装官方长期维护的SDK包，执行命令pip install azure-data-tables azure-storage-blob azure-identity
• 权限校准：
  • 源存储账户：给执行备份的身份（托管身份/服务主体/本地登录账号）分配Storage Table Data Reader角色，仅授予表数据的读权限
  • 目标存储账户：给同一身份分配Storage Blob Data Contributor角色，授予Blob的读写权限
  • 注：之前仅配置Blob Data Contributor角色无法访问源端表存储，这是azCopy任务报权限错误的核心原因，不是工具本身的设计逻辑问题。权限校准完成后，azCopy也可以正常完成表导出操作。

核心实现流程 #

整个流程适配大规模表存储的备份场景，不存在全量内存加载的性能瓶颈：

• 按命名前缀筛选需要备份的表范围（比如所有以prod_开头的业务表），自动跳过不需要备份的系统表
• 分表逐次拉取实体：利用表存储SDK自带的分页能力，每次拉取最大1000条实体，自动处理分页续传令牌
• 数据落盘到Blob：以JSON Lines格式流式写入目标Blob，Blob的路径结构按{容器名}/{备份批次标识}/{表名}.jsonl组织，避免单文件过大
• 断点续传支持：每完成一个表的全量备份，就在目标容器下的_backup_marker路径写入对应表的完成状态，下次执行时自动跳过已备份完成的表

可复用核心代码片段 #

from azure.identity import DefaultAzureCredential
from azure.data.tables import TableServiceClient
from azure.storage.blob import BlobServiceClient
import datetime
import json

# 基础配置
SOURCE_STORAGE_ACCOUNT = "your-source-table-storage"
TARGET_STORAGE_ACCOUNT = "your-target-blob-storage"
TARGET_BACKUP_CONTAINER = "table-backups"
TABLE_NAME_PREFIX = "prod_" # 仅备份符合该前缀的表
BATCH_SIZE = 500

# 使用Azure AD身份认证，无需硬编码存储密钥
credential = DefaultAzureCredential()
# 初始化服务客户端
table_service = TableServiceClient(
    endpoint=f"https://{SOURCE_STORAGE_ACCOUNT}.table.core.windows.net",
    credential=credential
)
blob_service = BlobServiceClient(
    account_url=f"https://{TARGET_STORAGE_ACCOUNT}.blob.core.windows.net",
    credential=credential
)
backup_container = blob_service.get_container_client(TARGET_BACKUP_CONTAINER)
backup_container.create_container(exists_ok=True)

# 执行备份流程
backup_batch_id = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
for table in table_service.list_tables():
    if not table.name.startswith(TABLE_NAME_PREFIX):
        continue
    print(f"Start backup for table: {table.name}")
    table_client = table_service.get_table_client(table.name)
    blob_path = f"{backup_batch_id}/{table.name}.jsonl"
    blob_client = backup_container.get_blob_client(blob_path)

    entity_count = 0
    upload_buffer = []
    for entity in table_client.list_entities():
        upload_buffer.append(json.dumps(entity))
        # 达到批量阈值就上传
        if len(upload_buffer) >= BATCH_SIZE:
            blob_client.upload_blob("\n".join(upload_buffer) + "\n", overwrite=True)
            entity_count += len(upload_buffer)
            upload_buffer.clear()
    # 写入剩余缓冲内容
    if upload_buffer:
        blob_client.upload_blob("\n".join(upload_buffer) + "\n", overwrite=True)
        entity_count += len(upload_buffer)
    
    print(f"Finished backup for {table.name}, total {entity_count} entities saved")

生产环境优化提示 #

• 调整BATCH_SIZE参数：根据实际表实体的平均大小调整，保证单次上传的内容不超过Blob的单块上传限制（4MB），避免请求失败
• 合理加并发：表数量较多时，可以用线程池对不同表的备份任务做并行调度，控制并发数在5-10之间，避免触发表存储的请求速率限制
• 自动清理过期备份：在脚本中加入逻辑，保留最近N天的备份文件，自动删除过老的备份以节省Blob存储成本
• 增加备份校验：每完成一个表的备份，拉取Blob文件的大小和表的实体总数做比对，避免静默失败导致备份残缺

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