モデルが更新されたときにジョブをトリガーする

モデル更新トリガーを使用して、 Unity カタログでモデルの変更が発生したときにジョブを自動的に実行します。 この機能は、モデルの更新を追跡するための cron スケジュールまたは永続的なクラスターの必要性に代わる機能です。

Important

モデルの更新トリガーは ベータ版です。

モデルの更新トリガーは、次の 2 つの主要なペルソナ用に設計されています。

  • データ サイエンティスト は、新しいモデル バージョンの準備ができたり、エイリアスが設定されている場合、1 つのモデルで、または一度に複数のモデルにわたってジョブをトリガーしたりできます。 たとえば、所有するモデルが変更されるたびに、検証、テスト、昇格のジョブを自動的に実行できます。 「例: 新しいバージョンまたはエイリアスでモデルを検証する」を参照してください。
  • 管理者は、 メタストア全体 (またはスキーマ) のすべてのモデルを監視できます。 たとえば、管理者は、作成されたすべての新しいモデルを自動的に監査できます。 例: スキーマまたはメタストア全体のモデルの監査を参照してください。

モデル更新トリガーの仕組み

モデル更新トリガーは、Unity カタログのスコープでモデル イベントを監視し、一致するイベントが発生したときにジョブの実行をトリガーします。 モデル更新トリガーを構成するときは、 スコープ と 条件を選択します。

使用可能なスコープは次のとおりです。

  • モデル: 1 つの登録済みモデル。
  • スキーマ: スキーマ内のすべてのモデル。
  • メタストア: メタストア内のすべてのモデル。 メタストア スコープのトリガーを構成できるのは、metastore 管理者だけです。

条件によって、実行をトリガーするタイミングが決まります。

  • モデルが作成される: 新しい登録済みモデルがスコープ内に作成されます。
  • モデル バージョンの準備ができました:新しいモデル バージョンがスコープ内で準備完了になります。
  • モデル エイリアスが設定されている: 指定したエイリアスがモデル バージョンに適用されます。 トリガーごとに最大 10 個のエイリアスを指定でき、いずれかのエイリアスが適用されるとトリガーが応答します。

モデルの更新トリガーでは、クラウド プロバイダーのコスト以外の追加コストは発生しません。

バッチ更新

モデルの更新では、1 分あたり約 1 回の新しいイベントのポーリングがトリガーされます。 各ポーリング間隔で検出された変更はバッチ処理され、ジョブ パラメーターとして次のジョブ実行に渡されます。 モデル更新トリガーに関連付けられているジョブ パラメーターを参照してください。

バッチは固定された数の更新を保持するわけではありません。 各実行には、ジョブ パラメーター値の 10,000 文字の制限内に収まるだけの更新が含まれます。 トリガーはバッチに1つずつ更新を追加し、制限を超える更新の前に停止するため、カウントは各シリアライズされた更新の長さに依存します。

パラメータ値はコンパクトJSONで、コンマで区切られた括弧付きのオブジェクトリストで、各オブジェクトにはモデルのフルネームが記載されており、モデルバージョン用のバージョンが 準備でき、 モデルエイリアス用のバージョンと別名の両方が設定されています。 これらの名前は制限を消費するため、短いモデル名やエイリアス名は各実行でより多くの更新の余裕を残します。

たとえば、main.ml_models.fraud_detection という名前のモデルのバージョン 123 に対する、エイリアス champion を持つ 1 つの [モデル エイリアスの設定] の更新は、84 文字にシリアル化されます:

{"full_name":"main.ml_models.fraud_detection","version":123,"alias_name":"champion"}

最初のアップデート以降は、前回のアップデートと区切るコンマに1文字のコストがかかり、括弧の部分は2文字かかります。 したがって、収まる更新数は、2 + 84n + (n - 1) が 10,000 以内に収まる最大数 n となります。つまり、合計 9,946 文字で 117 件の更新に相当します。

同じモデル名を使い、条件に含まれている場合は3桁のバージョンとエイリアスを用いて、各条件で同じ計算を繰り返します champion:

状態 更新あたりのキャラクター数 実行ごとの更新
モデルが作成される 46 212
モデル バージョンの準備ができました 60 163
モデルエイリアスは設定されます 84 117

長い名前ほどこれらの数字は下がります。 200文字のモデル名では、1回の実行でモデルが作成される更新が46回分、モデル バージョンの準備完了更新が43回分、モデル エイリアスが設定される更新が39回分発生します。 短いエイリアスを50文字のエイリアスに置き換えると、モデル エイリアスが設定されていますの更新回数は117回から78回に減ります。

1回の実行に収まる更新数を上回る更新が多い場合、トリガーは残りの更新を後の実行に1回のポーリング間隔ごとに1バッチで転送します。 その後の数分間は断続的な噴出が予想され、その後収まります。 スコープによってトリガーが一定期間処理できるよりも速く更新が生成された場合、トリガーは追い付かずに間隔ごとに実行され、最終的には失敗して停止します。 「持続的な高いイベント量」を参照してください。

継続的な高いイベント量

モデル更新トリガーは、平均すると更新の到着頻度が 1 回の実行で処理可能な量を上回らないスコープで最も効果的に機能します。 バッチ更新を参照してください。

増え続けるバックログから保護するために、ポーリング間隔ごとに約 1 時間連続して実行されるトリガーは失敗し、停止します。 エラーはトリガーで報告されます。 トリガーは自動的には回復せず、手動で回復するまでそれ以上の実行は開始されません。

トリガーを回復するには:

  1. アップデートの頻度を下げて、1回のランで追いつくようにしましょう。 たとえば、次の推奨事項で説明するように、トリガーのスコープを絞り込んだり、監視を複数のトリガーに分割したりします。 この手順をスキップすると、復旧後にトリガーが再度失敗します。
  2. トリガーを一時停止してから再開することで、リセットします。 一時停止すると、トリガーの累積状態がクリアされ、現在のポイントから再開されます。 [ジョブの詳細] ウィンドウの [スケジュールとトリガー] セクションを使用するか、pause_statusを [PAUSED] に設定し、[ジョブ API] でUNPAUSEDします。 「既存のトリガーを管理する」を参照してください。

高度なオプション(トリガー間の最小時間 や 「最後の変更後待機」)は実行作成の頻度を制御しますが、実行が処理する更新回数は変わりません。 アップデートが一貫してランの処理速度を超えて届く場合、この失敗を防ぐことはできません。

大規模な対象範囲を確実に監視するには:

  • 各トリガーのスコープを絞り込みます。 1 つのメタストア スコープ トリガーではなく、スキーマ またはモデル スコープのトリガーを使用して、1 つの領域のバーストがメタストア全体の監視を停止しないようにします。
  • 監視を複数のトリガーとジョブに分割して、各トリガーが低いボリューム スコープを監視できるようにします。
  • トリガーされたジョブを軽量に保ち、For each タスクを使用して各更新を処理します。 For each タスクを使用したモデル更新の処理を参照してください。
  • 広範な範囲にわたる継続的な大量のボリュームについては、Azure Databricks アカウント チームに連絡してオプションについて話し合ってください。

始める前の準備

モデル更新トリガーを使用するには、次のものが必要です。

  • ワークスペースで Unity カタログが有効になっている必要があります。
  • トリガーを監視するターゲット モデルまたはスキーマに対する EXECUTE 特権が必要です。 EXECUTEしないと、それらのモデルへの読み取りと書き込みが失敗する可能性があります。
  • メタストア スコープトリガーを構成するには、メタストア管理者である必要があります。また、ジョブが Unity カタログ内の任意のモデルにアクセスできるように、メタストア内のすべての現在および将来のカタログに対する EXECUTE 権限を自分自身に付与する必要があります。

モデル更新トリガーを追加する

既存のジョブにモデル更新トリガーを追加するには:

  1. Azure Databricksワークスペースのサイドバーで、Jobs & Pipelinesをクリックします。
  2. ジョブの一覧で、トリガーを追加するジョブの名前をクリックします。
  3. 右側の [ジョブの詳細 ] ウィンドウで、[ トリガーの追加] をクリックします。
  4. [トリガーの種類] で、[モデルの更新] を選択します。
  5. [ スコープ] で[ モデル]、[ スキーマ]、または [メタストア] を選択し、監視するモデルまたはスキーマを指定します。
  6. [ 条件] で、トリガーするイベントを選択します。
    • モデルが作成される
    • モデル バージョンの準備ができました
    • モデルのエイリアスが設定されています。 監視するエイリアスを最大 10 個指定します。 トリガーは、指定されたエイリアスのいずれかが設定されると応答します。
  7. (省略可能) 詳細オプションを構成します。
    • トリガー間の最小時間 (秒単位): 前回の実行後に実行をトリガーするまでの最小時間。 この期間中に更新されたモデルは、待機時間の経過後にのみ実行をトリガーします。 この設定は、実行の頻度を制御するために使用します。
    • 最後の変更後の待機時間 (秒単位): モデルの更新後に実行をトリガーするまでの待ち時間。 この期間中に別のモデルを更新すると、タイマーがリセットされます。 この設定は、モデルの更新がバッチで行われ、すべての更新が到着した後にバッチ全体を処理する必要がある場合に使用できます。
  8. 構成を検証するには、[ テスト トリガー] をクリックします。 エラーがない場合は、ボタンに [成功] と表示されます。
  9. 保存 をクリックします。

トリガーを保存した後、トリガーが初期化されるまで約 1 分待ちます。 メッセージ トリガーは間もなく評価され、初期化が進行中であることを示します。 初期化後、スコープ内の一致するモデル イベントによってジョブの実行がトリガーされます。

このトリガーを後で編集、一時停止、または削除するには、[ジョブの詳細] ウィンドウの [スケジュールとトリガー] セクションを使用します。 「既存のトリガーを管理する」を参照してください。

注

ジョブ API からモデル更新トリガーを構成することもできます。 ジョブ API を使用したモデル更新トリガーの構成を参照してください。

例: 新しいバージョンまたはエイリアスでモデルを検証する

モデルが変更されたときに検証、テスト、または昇格ジョブを実行するには、モデルスコープと、モデル バージョンの準備完了またはモデル エイリアスが設定されている条件のいずれかを指定して、トリガーを構成します。 エイリアスの変更については、 prod や stagingなど、監視するエイリアスを指定します。 一度に複数のモデルを監視するには、代わりに スキーマ スコープを使用します。 トリガーが初期化されると、監視対象モデルに一致する変更によってジョブの実行がトリガーされます。

例: スキーマまたはメタストア全体のモデルを監査する

スキーマまたはメタストアで作成されたすべてのモデルを監査するには、 スキーマ または メタストア スコープを使用してトリガーを構成し、 モデルが作成条件になります 。 トリガーが初期化されると、スコープ内に作成されたすべてのモデルによって、変更を記録または検証できるジョブ実行がトリガーされます。

モデル更新トリガーに関連付けられているジョブ パラメーター

モデル更新トリガーが起動すると、実行をトリガーした変更に関する情報は、 {{job.trigger.model.updates}} 動的値参照を介してジョブ実行で使用できます。 値は、バッチ内のモデルの更新の JSON リストです。

[
  {
    "full_name": "model.full.name1",
    "version": 123,
    "alias_name": "prod"
  }
]

フィールドは次のように設定されます。

  • full_name: 作成または更新されたモデルの名前が常に設定されます。
  • version: モデル バージョンの準備完了イベントでは、準備完了となったバージョンが、モデルエイリアスが設定されたイベントでは、エイリアスが設定されたバージョンがそれぞれ設定されます。
  • alias_name: モデルのエイリアスが設定されましたのイベントが発生した場合にのみ、設定されたエイリアスの名前が入力されます。

たとえば、各条件のパラメーター値は次のようになります。

  • モデルが作成されます。

    [{ "full_name": "model.number.one" }, { "full_name": "model.number.two" }]

  • モデル バージョンの準備ができました。

    [
  { "full_name": "model.number.one", "version": 7 },
  { "full_name": "model.number.two", "version": 3 }
]

  • モデルのエイリアスが設定されています。

    [
  { "full_name": "model.number.one", "version": 7, "alias_name": "prod" },
  { "full_name": "model.number.two", "version": 3, "alias_name": "staging" }
]

タスクで更新を使用できるようにするには、値が {{job.trigger.model.updates}}ジョブ パラメーターを追加します。 パラメーターを追加するには、ジョブのサイド パネルで [ パラメーターの編集 ] をクリックし、パラメーター値を {{job.trigger.model.updates}} に設定します。 ジョブ パラメーターの詳細については、「ジョブの パラメーター化」を参照してください。

次のノートブックは、 events という名前のパラメーターを読み取り、実行に渡されたモデルの更新を処理します。

import json

json_list = dbutils.widgets.get("events")
data = json.loads(json_list)

for item in data:
    print(f"Full Name: {item['full_name']}, Version: {item.get('version')}, Alias Name: {item.get('alias_name')}")

For each タスクを使用してモデルの更新を処理する

各タスク は、リスト内の各要素に対して入れ子になったタスクを 1 回ずつ実行します。 モデル更新トリガーの場合、 {{job.trigger.model.updates}}の要素ごとに 1 つのイテレーションを実行できます。

  1. For each タスクが入力として{{job.trigger.model.updates}}を受け取るように構成します。
  2. 反復によって提供される各入力の値を読み取る入れ子になったタスクを構成します。
  3. 各更新プログラムの値は、入れ子になったタスクのウィジェット パラメーターとして使用できます。

次のノートブックは、ネストされたタスク内の各反復の値を読み込みます。

full_name = dbutils.widgets.get("full_name")
version = dbutils.widgets.get("version")
alias_name = dbutils.widgets.get("alias_name")

print(f"Full Name: {full_name}, Version: {version}, Alias Name: {alias_name}")

ジョブ API を使用してモデル更新トリガーを構成する

モデル更新トリガーを構成するには、trigger、jobs/create、またはjobs/update操作にjobs/reset オブジェクトを追加します。 jobs/update の使用例を次に示します。

{
  "job_id": 574587036927544,
  "new_settings": {
    "trigger": {
      "pause_status": "UNPAUSED",
      "model": {
        "securable_name": "main.default",
        "condition": "MODEL_ALIAS_SET",
        "aliases": ["alias1", "alias2"],
        "min_time_between_triggers_seconds": 3600,
        "wait_after_last_change_seconds": 120
      }
    },
    "parameters": [
      {
        "default": "{{job.trigger.model.updates}}",
        "name": "events"
      }
    ]
  }
}

model トリガー オブジェクトの場合:

  • securable_name: 監視するスキーマまたはモデル。 メタストア全体を監視するには、空のままにするか省略します。
  • condition: MODEL_CREATED、 MODEL_VERSION_READY、または MODEL_ALIAS_SETのいずれか。
  • aliases: 監視するエイリアス。 MODEL_ALIAS_SET条件にのみ適用されます。

失敗したモデル更新トリガーの通知を受信する

モデル更新トリガーの評価に失敗した場合に通知を受け取るために、ジョブの失敗時に電子メールまたはシステムの送信先通知を構成します。 「ジョブについての通知を追加する」をご覧ください。

制限事項

モデルの更新トリガーには、次の制限があります。

  • ワークスペースごとに最大 100 個のモデル更新トリガーを構成できます。 この制限は、ケース バイ ケースで引き上げられます。
  • 1回のジョブ実行は、ジョブパラメータ値の10,000文字制限(通常100文字を超える)に収まる限りのモデル更新を含みます。 追加の保留中の更新は後の実行で1分あたり1バッチで処理されます。 バッチ更新を参照してください。
  • 各モデル更新トリガーでは、最大 10 個のエイリアスを監視できます。
  • トリガーがポーリング間隔ごとに約 1 時間連続して実行されると、エラーで失敗し、実行のトリガーが停止します。 自動的には回復しません。更新の速度を下げてから、一時停止して一時停止を解除してトリガーをリセットする必要があります。 これは、スコープが更新をトリガーが持続的な期間処理できるよりも速く生成する場合に発生します。 「持続的な高いイベント量」を参照してください。
  • メタストア スコープのトリガーを構成するには、メタストア管理者である必要があります。また、ジョブが任意のモデルにアクセスするには、Unity カタログのすべての現在および将来のカタログに対する EXECUTE 特権を自分自身に付与する必要があります。

FAQ

デプロイ ジョブとモデルの更新トリガーを使用する必要があるタイミング

次の必要がある場合は、モデル更新トリガーを使用します。

  • モデル作成イベントまたはエイリアスの変更を監視します。
  • スキーマやメタストアなど、1 つのモデルよりも広い範囲でイベントを監視します。
  • モデルごとに個別のジョブを構成する代わりに、1 つのトリガーで多くのモデルを監視します。

モデル バージョンの作成とジョブ実行の関係に関する厳密な UI 結合とアクティビティ ログが必要な場合は、 デプロイ ジョブを使用します。

  • Last updated on