Azure ML パイプラインの実行ステータスを取得してみた

やりたいこと

MLパイプラインを公開して外部からパイプラインを動かすときに、実行ステータスを確認したい。

具体的には、Azure ML Workspace で実行されている最新のジョブの状態（実行中、完了、失敗など）を API 経由で取得できるようにしたい。

パイプラインの開始とジョブの取り消しは別記事の予定。

前提条件

• Azure ML Workspace、Pipeline が作成済み
• Python 環境（Flask を使用）
• Azure CLI でログイン済み

必要なライブラリのインストール

uv add azure-ai-ml azure-identity

環境変数の設定

.env ファイルに Azure ML の接続情報を追加します。

AZURE_SUBSCRIPTION_ID=your-subscription-id
AZURE_RESOURCE_GROUP=your-resource-group
AZURE_WORKSPACE_NAME=your-workspace-name

実装方法

1. Azure ML Client を取得する関数を作成

まずは Azure ML に接続するためのクライアントを取得する関数を作ります。

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

def get_ml_client() -> MLClient:
    """Azure ML Clientを取得"""
    subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
    resource_group = os.getenv("AZURE_RESOURCE_GROUP")
    workspace_name = os.getenv("AZURE_WORKSPACE_NAME")

    credential = DefaultAzureCredential()
    return MLClient(
        credential=credential,
        subscription_id=subscription_id,
        resource_group_name=resource_group,
        workspace_name=workspace_name,
    )

DefaultAzureCredential() は、ローカル開発時は Azure CLI の認証情報を使い、本番環境ではマネージドIDを自動的に使用してくれる便利な認証方法です。

2. 最新ジョブを取得する関数を作成

次に、最新のジョブ情報を取得する関数を作ります。

def get_latest_job():
    """最新のジョブ情報を取得"""
    ml_client = get_ml_client()

    # 最新のジョブを1件取得（ジェネレータで返される）
    jobs = ml_client.jobs.list(max_results=1)

    latest_job = None
    for job in jobs:
        latest_job = job
        break  # 1件取得したらループを抜ける

    if not latest_job:
        return None

    # ジョブ情報を辞書形式で返す
    return {
        'job_name': latest_job.name,
        'status': latest_job.status,
        'creation_time': str(latest_job.creation_context.created_at) if latest_job.creation_context else None,
        'error': str(latest_job.properties.get('error')) if hasattr(latest_job, 'properties') and latest_job.properties.get('error') else None
    }

ここでのポイントは、ml_client.jobs.list(max_results=1) がジェネレータを返すことです。そのため、for ループで取得する必要があります。

また、jobs.list() は作成日時の降順でソートされているため、max_results=1 で最新の1件だけを取得できます。

3. Flask のエンドポイントを作成

最後に、API エンドポイントを作成します。

from flask import Blueprint, jsonify

bp = Blueprint('azure_ml', __name__)

@bp.route('/azure-ml/jobs/latest', methods=['GET'])
def get_latest_job_status():
    """Azure MLの最新ジョブ状態を取得"""
    try:
        job_info = get_latest_job()

        if not job_info:
            return jsonify({
                "message": "ジョブが見つかりません"
            }), 404

        return jsonify(job_info), 200

    except Exception as e:
        return jsonify({
            "message": "An internal server error has occurred"
        }), 500

実際に呼び出してみる

curl http://localhost:5000/azure-ml/jobs/latest

レスポンス例：

{
  "job_name": "my-training-job-20250105",
  "status": "Completed",
  "creation_time": "2025-01-05 10:30:00",
  "error": null
}

取得できるステータスの種類

Azure ML のジョブには以下のようなステータスがあります。

ステータス	説明
Running	実行中
Completed	完了
Failed	失敗
Queued	キューに入っている
Starting	開始中
Canceled	キャンセル済み

https://learn.microsoft.com/ja-jp/rest/api/azureml/jobs/list?view=rest-azureml-2025-09-01&tabs=HTTP#jobstatus

どんな時に使える？

• MLパイプラインを外部から実行した後、完了を待ちたいとき
• ジョブの実行状況を監視ダッシュボードに表示したいとき
• CI/CD パイプラインで ML モデルの再学習状況を確認したいとき
• Slack や Teams に実行状況を通知したいとき

注意点

• DefaultAzureCredential を使う場合、ローカル開発では事前に az login が必要
• 本番環境ではマネージドIDの権限設定が必要（Azure ML Workspace への読み取り権限）
• ジェネレータから取得する際、リスト全体を展開すると大量のジョブが取得されるため、max_results を指定するのが望ましい

まとめ

• Azure ML Python SDK を使えば、簡単にジョブのステータスを取得できる
• MLClient.jobs.list() はジェネレータを返すため、for ループでの取得が必要
• DefaultAzureCredential を使うことで、環境ごとに認証方法を切り替えられる
• 外部からパイプラインを実行する際の監視に便利

参考

• Machine Learning REST API reference