如何将Azure ML作业输出的DataFrame自动注册为数据资产

方法1：通过Output参数自动注册（推荐） #

你当前代码中embeddings输出的path指定了已有的数据资产版本（azureml:embeddings:1），这会覆盖该版本内容而非自动创建新资产。要实现自动注册，只需替换path参数为register_with_name，并可添加可选元数据：

fetch_model_component = command(
    name="fetch_pre_trained_model_and_create_embeddings",
    display_name="Fetch Pre-Trained Model and create embeddings",
    description="fetches a pre-trained sbert model, and uses text to create document embeddings",

    inputs={
        "data": Input(type="uri_folder"),
        "registered_model_name": Input(type="string")
    },
    outputs={
        "model": Output(type="uri_folder", mode="rw_mount"),
        "embeddings": Output(
            type="uri_folder",
            mode="rw_mount",
            register_with_name="embeddings",  # 指定资产名称，自动生成新版本
            description="Document embeddings generated from SBERT model",
            tags={"task": "embedding-generation", "framework": "sbert"}
        )
    },
    code=train_src_dir,
    command="""python pre_trained.py \
            --data ${{inputs.data}} --registered_model_name ${{inputs.registered_model_name}}
            --model ${{outputs.model}} --embeddings ${{outputs.embeddings}} \
            """,
    environment=f"{pipeline_job_env.name}:{pipeline_job_env.version}",
)

关键说明： #

• register_with_name：指定数据资产的固定名称，管道每次运行后会自动将输出文件夹注册为该资产的新版本。
• 你现有的CSV保存代码df.to_csv(os.path.join(args.embeddings, "embeddings.csv"), index=False)无需修改，输出文件夹包含的CSV会被纳入注册的资产中。
• 若需仅注册单个CSV文件而非整个文件夹，可将type改为uri_file，并调整保存路径直接指向文件。

方法2：添加独立的注册步骤（精细控制） #

如果需要更灵活的配置（比如仅注册CSV文件、添加自定义元数据），可以新增专门的组件完成数据资产注册：

步骤1：编写注册脚本（register_embeddings.py） #

from azure.ai.ml import MLClient
from azure.ai.ml.entities import Data
from azure.identity import DefaultAzureCredential
import argparse

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--embeddings_file", type=str, required=True)
    parser.add_argument("--asset_name", type=str, required=True)
    args = parser.parse_args()

    # 初始化ML客户端（自动获取当前工作区信息）
    ml_client = MLClient(DefaultAzureCredential())

    # 定义数据资产
    embeddings_asset = Data(
        path=args.embeddings_file,
        type="uri_file",
        name=args.asset_name,
        description="Document embeddings in CSV format",
        tags={"data_format": "csv", "model_source": "sbert"}
    )

    # 创建或更新数据资产
    ml_client.data.create_or_update(embeddings_asset)

if __name__ == "__main__":
    main()

步骤2：在管道中添加注册组件 #

# 定义注册组件
register_embeddings_component = command(
    name="register_embeddings_asset",
    display_name="Register Embeddings CSV as Data Asset",
    inputs={
        "embeddings_file": Input(
            type="uri_file",
            path=f"{fetch_model_component.outputs.embeddings}/embeddings.csv"
        ),
        "asset_name": Input(type="string", default="embeddings")
    },
    code="./registration_src",  # 存放register_embeddings.py的文件夹路径
    command="python register_embeddings.py --embeddings_file ${{inputs.embeddings_file}} --asset_name ${{inputs.asset_name}}",
    environment=f"{pipeline_job_env.name}:{pipeline_job_env.version}",
)

# 将注册组件加入管道，依赖生成embeddings的步骤
pipeline_job = pipeline(
    jobs={
        "generate_embeddings": fetch_model_component,
        "register_embeddings": register_embeddings_component
    }
)

关键说明： #

• 确保环境中安装了azure-ai-ml和azure-identity包，可在环境的conda.yml或pip依赖中添加。
• 此方法允许单独控制注册逻辑，比如针对CSV文件设置特定元数据或资产类型。

注意事项 #

• 不要同时在Output中指定path和register_with_name，二者会冲突（path指向已有资产，register_with_name创建新资产）。
• 使用DefaultAzureCredential时，确保管道运行的身份（如计算集群的托管标识）拥有数据资产的创建权限。

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