```python
# Global var: desired_status (change from "Completed" to "Failed" according to your requirements)
desired_status = "Completed"
# --------------------------------------------------------------------------------------------------
# Connect to AML and set tracking URI in mlflow
# --------------------------------------------------------------------------------------------------
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azureml.core.experiment import Experiment
from azureml.core.workspace import Workspace
# Connect to AML
client = MLClient(
credential= InteractiveBrowserCredential(),
subscription_id="my-subscription-id",
resource_group_name="my-resource-group",
workspace_name="my-workspace"
)
# --------------------------------------------------------------------------------------------------
# Get workspace
# --------------------------------------------------------------------------------------------------
workspace = Workspace(
client.subscription_id,
client.resource_group_name,
client.workspace_name
)
# --------------------------------------------------------------------------------------------------
# Retrieve schedules
# --------------------------------------------------------------------------------------------------
schedules = client.schedules.list()
# optional: filter those with desired name patterns
selected_schedules = [
schedule
for schedule in schedules
if "inference" in schedule.name
]
# --------------------------------------------------------------------------------------------------
Select schedules *names* that meet the criteria (at least one run completed)
# --------------------------------------------------------------------------------------------------
finished_schedules = []
for schedule in selected_schedules:
experiment = Experiment(workspace, schedule.create_job.experiment_name)
if any(map(lambda x: x.status == desired_status, experiment.get_runs())):
finished_schedules.append(schedule.name)
```
If we want the ones that have their last run failed:
```python
last_run_failed_schedules = []
times = []
for schedule in selected_schedules:
experiment = Experiment(workspace, schedule.create_job.experiment_name)
last_run = next(experiment.get_runs())
if last_run.status == "Failed":
last_run_failed_schedules.append(schedule.name)
```
First code snippet, slow if we have many experiments and runs (see below second code snippet):
```
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
# Connect to AML
client = MLClient(
credential= InteractiveBrowserCredential(),
subscription_id="my-subscription-id",
resource_group_name="my-resource-group",
workspace_name="my-workspace"
)
# set tracking uri if run locally
mlflow_tracking_uri = client.workspaces.get(client.workspace_name).mlflow_tracking_uri
mlflow.set_tracking_uri(mlflow_tracking_uri)
# select all runs
# (slow, see alternative in second code snippet below)
all_runs = mlflow.search_runs(search_all_experiments=True)
# select runs whose my-metric-name is bigger than my_threshold
metric_name = "my-metric-name"
my_threshold = 0.9
selected_runs = all_runs.loc[all_runs[f"metrics.{metric_name}"] > my_threshold]
# get corresponding experiment IDs:
selected_experiment_ids = selected_runs["experiment_id"].unique()
# retrieve all experiments
# (slow, see alternative in second code snippet below)
exps = mlflow.search_experiments()
# show experiments whose experiment-id matches the selected runs
for exp in exps:
if exp.experiment_id in selected_experiment_ids:
print(f"Experiment name: {exp.name}\nExperiment ID: {exp.experiment_id}")
```
If we can narrow down our list of candidate experiments using either a list of experiment names we want to focus on, the user name who created the experiments, or both, we can get a faster response. This would be the resulting code after setting the tracking uri:
```
# restrict experiment name to be one of the following:
experiment_names = ["my-first-experiment", "my-second-experiment", "my-third-experiment"]
# restrict user id to be the following:
user_id = "my-user-id"
# select runs
all_runs = mlflow.search_runs(
experiment_names=experiment_names,
filter_string=f"tags.mlflow.user='{user_id}'",
)
# select runs whose my-metric-name is bigger than my_threshold
metric_name = "my-metric-name"
my_threshold = 0.9
selected_runs = all_runs.loc[all_runs[f"metrics.{metric_name}"] > my_threshold]
# get corresponding experiment IDs:
selected_experiment_ids = selected_runs["experiment_id"].unique()
# retrieve all experiments whose name is in the experiment_names list
filter_string = " or ".join([f"(name = {x})" for x in experiment_names])
exps = mlflow.search_experiments(filter_string=filter_string)
# show experiments whose experiment-id matches the selected runs
for exp in exps:
if exp.experiment_id in selected_experiment_ids:
print(f"Experiment name: {exp.name}\nExperiment ID: {exp.experiment_id}")
```
