如何让已发布的AzureML训练管道获取新训练数据并更新数据资产

问题根源 #

你直接覆盖本地CSV后重新提交管道仍用旧数据，是因为首次创建管道时，本地文件被打包为管道的静态依赖快照，发布后的管道会复用该快照，不会主动检测本地文件的变更。要实现生产场景下自动获取新数据，必须通过**Azure ML数据资产（Data Asset）**来动态管理数据版本和引用。

解决方案步骤 #

1. 将本地CSV迁移为Azure ML数据资产 #

首先把CSV文件上传到Azure ML工作区的默认存储容器（或自定义存储），然后创建数据资产指向该文件：

from azure.ai.ml import MLClient
from azure.ai.ml.entities import DataAsset
from azure.identity import DefaultAzureCredential

# 初始化ML客户端
ml_client = MLClient(DefaultAzureCredential(), subscription_id="<你的订阅ID>", resource_group_name="<资源组名>", workspace_name="<工作区名>")

# 创建数据资产（指向存储中的CSV文件）
data_asset = DataAsset(
    name="training-csv-data",
    path="azureml://datastores/workspaceblobstore/paths/<存储容器中的CSV路径>",  # 替换为你的CSV存储路径
    type="uri_file"
)
ml_client.data.create_or_update(data_asset)

2. 管道组件中动态引用数据资产 #

修改管道的输入配置，直接引用数据资产名称（不指定版本），确保每次运行都拉取最新版本：

from azure.ai.ml import Input
from azure.ai.ml.dsl import pipeline

@pipeline()
def training_pipeline():
    # 引用数据资产作为输入，不指定版本则默认取最新版
    train_data = Input(type="uri_file", path="azureml:training-csv-data:latest")
    
    # 你的训练组件调用，使用train_data作为输入
    train_job = train_component(
        data=train_data,
        # 其他参数...
    )

3. 更新数据资产版本（新数据到来时） #

当新CSV覆盖存储中的旧文件后，创建数据资产的新版本（不要覆盖旧版本，确保可追溯）：

# 创建新版本数据资产，版本号可手动指定或让系统自动生成
new_data_asset = DataAsset(
    name="training-csv-data",
    path="azureml://datastores/workspaceblobstore/paths/<更新后的CSV路径>",
    type="uri_file",
    version="2"  # 每次更新递增版本号，或省略让系统自动生成
)
ml_client.data.create_or_update(new_data_asset)

4. 生产场景：自动触发管道运行 #

要实现新数据到来时自动触发管道，可通过以下方式：

• 用Azure Event Grid监听存储容器的文件更新事件（如BlobCreated或BlobUpdated）
• 事件触发后调用Azure ML管道端点的REST API，提交管道运行请求
• 也可在Azure ML工作区中设置管道触发器，关联存储账户的文件变更事件

关键注意事项 #

• 绝对不要直接用本地Notebook目录的文件作为管道输入，因为管道运行时会将本地文件打包成静态快照，发布后无法更新
• 始终通过Azure ML数据资产管理训练数据，利用版本化特性实现数据追溯和动态引用
• 若使用uri_folder类型的数据资产，当文件夹内文件更新时，可直接创建新版本指向该文件夹，无需单独处理每个文件

内容的提问来源于stack exchange，提问作者amit.s