如何使用Azure Python SDK v2检索至少有一次成功运行的AzureML计划

以下是实现步骤和代码示例，帮助你筛选出满足条件的计划：

1. 安装并导入依赖 #

确保已安装Azure ML SDK v2：

pip install azure-ai-ml

导入所需模块：

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

2. 连接到AzureML工作区 #

通过凭证认证连接到目标工作区：

# 替换为你的订阅ID、资源组名称和工作区名称
subscription_id = "your-subscription-id"
resource_group_name = "your-resource-group-name"
workspace_name = "your-workspace-name"

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

3. 筛选符合条件的计划 #

遍历所有计划，检查每个计划触发的作业中是否存在至少一次成功完成的记录：

qualified_schedules = []

# 获取工作区中所有计划
all_schedules = ml_client.schedules.list()

for schedule in all_schedules:
    # 过滤出当前计划触发的所有作业
    triggered_jobs = ml_client.jobs.list(
        filter=f"schedule_id eq '{schedule.id}'"
    )
    # 检查是否有至少一个作业成功完成
    has_successful_run = any(job.status == "Completed" for job in triggered_jobs)
    
    if has_successful_run:
        qualified_schedules.append(schedule)

# 输出结果
print(f"共找到 {len(qualified_schedules)} 个符合条件的计划：")
for sched in qualified_schedules:
    print(f"计划名称: {sched.name}, ID: {sched.id}, 当前状态: {sched.status}")

关键说明 #

• 权限要求：需要拥有AzureML工作区的Contributor或Reader以上权限，才能读取计划和作业数据
• 作业状态判断：AzureML v2中作业成功完成的状态为Completed，若需包含其他成功类状态，可根据实际需求调整判断条件
• 性能优化：如果工作区中计划和作业数量较多，可添加时间范围过滤（比如created_time ge '2024-01-01'）减少查询数据量
• 计划状态过滤：若仅需保留活跃计划，可在判断中添加and schedule.status == "Active"

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