如何通过Python SDK获取Azure Machine Learning工作区已发布管道及元数据

前置操作 #

• 安装AML v2 Python SDK依赖：
  pip install azure-ai-ml azure-identity

步骤1：连接AML工作区 #

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

# 替换为你的工作区对应参数
subscription_id = "你的订阅ID"
resource_group_name = "你的资源组名称"
workspace_name = "你的AML工作区名称"

ml_client = MLClient(
    credential=DefaultAzureCredential(),
    subscription_id=subscription_id,
    resource_group_name=resource_group_name,
    workspace_name=workspace_name
)

步骤2：获取所有已发布管道的元数据 #

调用ml_client.pipelines.list()接口可以拉取全量已发布管道，可获取的核心元数据包括：

• 管道唯一ID、显示名称、描述
• 管道版本、创建/更新时间戳
• 发布人信息、管道状态（启用/禁用）
• 管道调用端点、默认输入参数

示例代码：

# 获取所有已发布管道
published_pipelines = list(ml_client.pipelines.list())

# 遍历打印核心信息
for pipeline in published_pipelines:
    print(f"管道ID: {pipeline.id}")
    print(f"管道名称: {pipeline.name}")
    print(f"创建时间: {pipeline.creation_context.created_at}")
    print(f"发布人: {pipeline.creation_context.created_by}")
    print(f"管道状态: {pipeline.properties.get('pipeline_status', '未知')}")
    print("-"*50)

步骤3：查询管道关联的调度规则 #

调用ml_client.schedules.list()接口拉取所有调度规则，通过过滤pipeline_job_id字段关联到对应管道，可获取的调度元数据包括：

• 调度名称、描述、唯一ID
• 触发规则（cron表达式、重复频率、触发时间窗口）
• 调度状态（启用/禁用）、运行时参数
• 最近一次触发时间、下一次计划触发时间

示例代码：

# 拉取全量调度规则
all_schedules = list(ml_client.schedules.list())

# 按管道ID分组关联调度
pipeline_schedule_map = {}
for schedule in all_schedules:
    # 仅筛选管道调度类型的规则
    if hasattr(schedule, 'job') and schedule.job.type == 'pipeline':
        pipeline_id = schedule.job.job_id
        if pipeline_id not in pipeline_schedule_map:
            pipeline_schedule_map[pipeline_id] = []
        pipeline_schedule_map[pipeline_id].append({
            "调度名称": schedule.name,
            "触发规则": schedule.trigger,
            "是否启用": schedule.is_enabled,
            "创建时间": schedule.creation_context.created_at
        })

# 打印每个管道对应的调度信息
for pipeline in published_pipelines:
    print(f"管道名称：{pipeline.name} 对应调度规则：")
    schedules = pipeline_schedule_map.get(pipeline.id, [])
    if not schedules:
        print("无关联调度规则")
    else:
        for s in schedules:
            print(f"- 调度名：{s['调度名称']}，启用状态：{s['是否启用']}，触发规则：{s['触发规则']}")
    print("-"*50)

内容的提问来源于stack exchange，提问作者Lukas Lötters