如何通过Azure Python SDK for ADF引用存储账户中JSON定义动态创建数据流？

你不能直接通过指定存储账户中JSON文件路径的方式调用adf_client.data_flows.create_or_update()，因为该方法要求传入的是DataFlowResource对象，而非文件路径。正确的做法是先读取存储中的JSON定义，再转换成方法所需的格式后调用接口，具体步骤如下：

步骤说明 #

• 1. 使用Azure存储SDK读取存储账户中的数据流JSON定义文件
• 2. 将JSON内容反序列化为Python字典或DataFlowResource对象
• 3. 调用ADF SDK的create_or_update方法传入该对象完成创建

示例代码 #

from azure.storage.blob import BlobServiceClient
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import DataFlowResource
import json

# 读取存储账户中的数据流JSON
storage_conn_str = "<你的存储账户连接字符串>"
container_name = "<存储容器名>"
blob_name = "<数据流JSON文件名>"

blob_service_client = BlobServiceClient.from_connection_string(storage_conn_str)
blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
dataflow_json = json.loads(blob_client.download_blob().readall().decode("utf-8"))

# 初始化ADF客户端
adf_client = DataFactoryManagementClient("<你的Azure凭据>", "<订阅ID>")
resource_group = "<资源组名称>"
factory_name = "<ADF工厂名称>"
dataflow_name = "<目标数据流名称>"

# 创建并提交数据流资源
dataflow_resource = DataFlowResource(properties=dataflow_json["properties"])
result = adf_client.data_flows.create_or_update(
    resource_group_name=resource_group,
    factory_name=factory_name,
    data_flow_name=dataflow_name,
    data_flow=dataflow_resource
)

print(f"数据流操作完成: {result.name}")

关键注意事项 #

• 确保存储账户的访问权限配置正确，ADF SDK所在环境能读取到目标JSON文件
• JSON定义需符合ADF数据流的官方结构规范，避免因格式错误导致创建失败

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