Azure Data Factory如何根据JSON配置复制Blob存储文件并压缩容器

核心逻辑 #

你只需要完成JSON解析、遍历执行复制两步即可，全流程走Blob服务端接口无需下载文件到本地：

• 先把你读取到的JSON配置文本反序列化为数组对象，每个元素对应一条复制规则
• 遍历数组逐行读取from（源Blob路径）和to（目标Blob路径），调用对应云厂商Blob存储的复制接口完成单文件迁移
• 所有文件复制完成后再对目标路径执行压缩操作即可

代码示例（Python + Azure Blob Storage） #

如果你用的是其他厂商的对象存储（OSS/COS等），只需替换对应的Blob操作SDK接口即可，逻辑完全一致：

import json
from azure.storage.blob import BlobServiceClient

# 基础配置替换为你自己的存储信息
BLOB_CONN_STR = "你的Blob存储连接字符串"
CONTAINER_NAME = "你的Blob容器名"

# 初始化存储客户端
blob_svc_client = BlobServiceClient.from_connection_string(BLOB_CONN_STR)

# 此处替换为你实际读取到的JSON配置文本
config_content = """
[
  {
    "from": "00000000-0000-0000-0000-000000000001/1.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000001/1.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/2.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000002/2.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/3.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000003/3.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/4.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000004/4.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/5.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000001/5.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/6.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000002/6.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/7.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000003/7.txt"
  },
  {
    "from": "00000000-0000-0000-0000-000000000001/8.txt",
    "to": "00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000004/8.txt"
  }
]
"""

# 解析JSON为可遍历的规则列表
copy_rules = json.loads(config_content)

# 批量执行复制
for rule in copy_rules:
    source_blob = blob_svc_client.get_blob_client(container=CONTAINER_NAME, blob=rule["from"])
    target_blob = blob_svc_client.get_blob_client(container=CONTAINER_NAME, blob=rule["to"])
    # 触发服务端异步复制，大文件可额外加轮询复制状态的逻辑
    target_blob.start_copy_from_url(source_blob.url)

注意事项 #

如果你的文件体积较大，建议在复制后增加判断复制状态的逻辑，确认所有文件复制完成后再执行压缩操作，避免出现压缩包文件不全的问题。

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