Auslösen von Machine Learning-Pipelines

2025-06-24

GILT FÜR:Azure Machine Learning SDK v1 für Python

Wichtig

Dieser Artikel enthält Informationen zur Verwendung des Azure Machine Learning SDK v1. SDK v1 ist ab dem 31. März 2025 veraltet. Der Support für sie endet am 30. Juni 2026. Sie können SDK v1 bis zu diesem Datum installieren und verwenden.

Es wird empfohlen, vor dem 30. Juni 2026 zum SDK v2 zu wechseln. Weitere Informationen zu SDK v2 finden Sie unter Was ist Azure Machine Learning CLI und Python SDK v2? und die SDK v2-Referenz.

In diesem Artikel erfahren Sie, wie Sie programmgesteuert eine Pipeline für die Ausführung in Azure planen. Sie können einen Plan erstellen, der auf verstrichener Zeit oder auf Änderungen im Dateisystem beruht. Sie können zeitbasierte Zeitpläne verwenden, um Routineaufgaben auszuführen, z. B. die Überwachung auf Datenabweichungen. Sie können änderungsbasierte Zeitpläne verwenden, um auf unregelmäßige oder unvorhersehbare Änderungen zu reagieren, z. B. neue Daten, die hochgeladen werden, oder alte Daten, die bearbeitet werden.

Nachdem Sie erfahren haben, wie Sie Zeitpläne erstellen, erfahren Sie, wie Sie sie abrufen und deaktivieren. Schließlich erfahren Sie, wie Sie andere Azure-Dienste, Azure Logic Apps und Azure Data Factory verwenden, um Pipelines auszuführen. Eine Logik-App ermöglicht komplexere Triggerlogik oder -verhalten. Mit Azure Data Factory-Pipelines können Sie eine Machine Learning-Pipeline als Teil einer größeren Datenorchestrierung aufrufen.

Voraussetzungen

Ein Azure-Abonnement. Wenn Sie kein Azure-Abonnement besitzen, können Sie ein kostenloses Konto erstellen.
Eine Python-Umgebung, in der das Azure Machine Learning SDK für Python installiert ist. Weitere Informationen finden Sie unter Erstellen und Verwalten wiederverwendbarer Umgebungen für Schulungen und Bereitstellungen mit Azure Machine Learning.
Ein Machine Learning-Arbeitsbereich mit einer veröffentlichten Pipeline. Sie können die in Create and run machine learning pipelines with Azure Machine Learning SDK erstellte verwenden.

Erforderliche Werte abrufen

Um eine Pipeline zu planen, benötigen Sie einen Verweis auf Ihren Arbeitsbereich, den Bezeichner Ihrer veröffentlichten Pipeline und den Namen des Experiments, in dem Sie den Zeitplan erstellen möchten. Sie können diese Werte mithilfe des folgenden Codes abrufen:

import azureml.core
from azureml.core import Workspace
from azureml.pipeline.core import Pipeline, PublishedPipeline
from azureml.core.experiment import Experiment

ws = Workspace.from_config()

experiments = Experiment.list(ws)
for experiment in experiments:
    print(experiment.name)

published_pipelines = PublishedPipeline.list(ws)
for published_pipeline in  published_pipelines:
    print(f"{published_pipeline.name},'{published_pipeline.id}'")

experiment_name = "MyExperiment" 
pipeline_id = "aaaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"

Erstellen eines Zeitplans

Zum Ausführen einer Pipeline auf wiederkehrender Basis erstellen Sie einen Zeitplan. Ein Schedule verknüpft eine Pipeline, ein Experiment und einen Auslöser. Der Auslöser kann entweder ein ScheduleRecurrence Wert sein, der die Wartezeit zwischen Aufträgen oder einen Datenspeicherpfad definiert, der ein Verzeichnis angibt, das auf Änderungen überwacht werden soll. In beiden Fällen benötigen Sie den Pipelinebezeichner und den Namen des Experiments, in dem der Zeitplan erstellt werden soll.

Importieren Sie am Anfang der Python-Datei die Klassen Schedule und ScheduleRecurrence:


from azureml.pipeline.core.schedule import ScheduleRecurrence, Schedule

Erstellen eines zeitbasierten Plans

Der ScheduleRecurrence Konstruktor verfügt über ein erforderliches frequency Argument, das auf eine der folgenden Zeichenfolgen festgelegt werden muss: "Minute", , "Hour", , "Day", "Week"oder "Month". Außerdem ist ein ganzzahliges interval Argument erforderlich, das angibt, wie viele frequency Einheiten zwischen den Startzeiten vergehen sollen. Optionale Argumente ermöglichen es Ihnen, spezifischere Informationen zu Startzeiten zu erhalten, wie in der ScheduleRecurrence-Dokumentation beschrieben.

Erstellen Sie einen Schedule, der alle 15 Minuten einen Auftrag startet:

recurrence = ScheduleRecurrence(frequency="Minute", interval=15)
recurring_schedule = Schedule.create(ws, name="MyRecurringSchedule", 
                            description="Based on time",
                            pipeline_id=pipeline_id, 
                            experiment_name=experiment_name, 
                            recurrence=recurrence)

Erstellen eines änderungsbasierten Plans

Pipelines, die durch Dateiänderungen ausgelöst werden, sind möglicherweise effizienter als zeitbasierte Zeitpläne. Wenn Sie vor dem Ändern einer Datei oder dem Hinzufügen einer neuen Datei zu einem Datenverzeichnis etwas unternehmen möchten, können Sie diese Datei vorverarbeiten. Sie können alle Änderungen an einem Datenspeicher oder Änderungen in einem bestimmten Verzeichnis innerhalb des Datenspeichers überwachen. Wenn Sie ein bestimmtes Verzeichnis überwachen, lösen Änderungen in Unterverzeichnissen dieses Verzeichnisses keinen Auftrag aus.

Hinweis

Änderungsbasierte Zeitpläne unterstützen nur die Überwachung von Azure Blob Storage.

Um eine Schedule dateireaktive Funktion zu erstellen, müssen Sie den Parameter datastore im Aufruf von Schedule.create festlegen. Zum Überwachen eines Ordners legen Sie das path_on_datastore-Argument fest.

Mit dem polling_interval Argument können Sie in Minuten die Häufigkeit angeben, mit der der Datenspeicher auf Änderungen überprüft wird.

Wenn die Pipeline mit DataPath PipelineParameter erstellt wurde, können Sie durch Festlegen des data_path_parameter_name-Arguments diese Variable auf den Namen der geänderten Datei einstellen.

datastore = Datastore(workspace=ws, name="workspaceblobstore")

reactive_schedule = Schedule.create(ws, name="MyReactiveSchedule", description="Based on input file change.",
                            pipeline_id=pipeline_id, experiment_name=experiment_name, datastore=datastore, data_path_parameter_name="input_data")

Optionale Argumente zum Erstellen eines Zeitplans

Zusätzlich zu den zuvor erläuterten Argumenten können Sie das status-Argument auf "Disabled" setzen, um einen inaktiven Zeitplan zu erstellen. Das continue_on_step_failure ermöglicht Ihnen, einen Booleschen Wert zu übergeben, der das standardmäßige Fehlverhalten der Pipeline überschreibt.

Anzeigen der geplanten Pipelines

Wechseln Sie in einem Browser zu Azure Machine Learning Studio. Wählen Sie im linken Bereich das Symbol "Endpunkte" aus. Wählen Sie im Bereich "Endpunkte " die Option "Endpunkte in Echtzeit" aus. Dadurch gelangen Sie zu einer Liste der Pipelines, die im Arbeitsbereich veröffentlicht sind.

Auf dieser Seite können Sie zusammenfassende Informationen zu allen Pipelines im Arbeitsbereich anzeigen: Namen, Beschreibungen, Status usw. Weitere Informationen erhalten Sie, indem Sie den Namen einer Pipeline auswählen. Auf der resultierenden Seite können Sie auch Informationen zu einzelnen Aufträgen abrufen.

Deaktivieren der Pipeline

Wenn Sie ein veröffentlichtes, aber nicht geplantes Pipeline Element haben, können Sie es mithilfe dieses Codes deaktivieren.

pipeline = PublishedPipeline.get(ws, id=pipeline_id)
pipeline.disable()

Wenn die Pipeline bereits geplant ist, müssen Sie zuerst den Plan abbrechen. Rufen Sie den Bezeichner des Zeitplans aus dem Portal ab, oder führen Sie diesen Code aus:

ss = Schedule.list(ws)
for s in ss:
    print(s)

Nachdem Sie den schedule_id Zeitplan haben, den Sie deaktivieren möchten, führen Sie diesen Code aus:

def stop_by_schedule_id(ws, schedule_id):
    s = next(s for s in Schedule.list(ws) if s.id == schedule_id)
    s.disable()
    return s

stop_by_schedule_id(ws, schedule_id)

Wenn Sie dann Schedule.list(ws) erneut ausführen, sollte eine leere Liste angezeigt werden.

Verwenden von Logik-Apps für komplexe Trigger

Sie können komplexere Triggerregeln oder -verhalten mithilfe von Logik-Apps erstellen.

Um eine Logik-App zum Auslösen einer Machine Learning-Pipeline zu verwenden, benötigen Sie den REST-Endpunkt für eine veröffentlichte Machine Learning-Pipeline. Erstellen und veröffentlichen Sie Ihre Pipeline. Suchen Sie anschließend mithilfe der Pipeline-ID den REST-Endpunkt Ihrer PublishedPipeline:

# You can find the pipeline ID in Azure Machine Learning studio

published_pipeline = PublishedPipeline.get(ws, id="<pipeline-id-here>")
published_pipeline.endpoint

Erstellen Sie eine Logik-App in Azure

Erstellen Sie nun eine Logik-App. Führen Sie nach der Bereitstellung Ihrer Logik-App die folgenden Schritte aus, um einen Auslöser für Ihre Pipeline zu konfigurieren:

Erstellen Sie eine vom System zugewiesene verwaltete Identität , um der App Zugriff auf Ihren Azure Machine Learning-Arbeitsbereich zu gewähren.
Wählen Sie im linken Bereich "Logik-App-Vorlagen " im Abschnitt "Entwicklungstools " aus.
Wählen Sie die Vorlage "Leerer Workflow" aus:
Wählen Sie im Designer "Trigger hinzufügen" aus.
Im Bereich "Trigger hinzufügen" suchen Sie nach blob. Wählen Sie den Trigger aus, wenn ein Blob hinzugefügt oder geändert wird (nur Eigenschaften).
Geben Sie im Bereich "Verbindung erstellen " die Verbindungsinformationen für das Blob Storage-Konto an, das Sie auf Blob-Ergänzungen oder -Änderungen überwachen möchten, und wählen Sie dann "Neu erstellen" aus. Wählen Sie den zu überwachenden Container aus.

Wählen Sie Intervall- und Häufigkeitswerte aus, die für Sie funktionieren.

Hinweis

Mit diesem Auslöser wird der ausgewählte Container überwacht, jedoch keine Unterordner.
Fügen Sie eine HTTP-Aktion hinzu, die ausgeführt wird, wenn ein Blob geändert wird oder ein neues Blob erkannt wird. Wählen Sie unter dem Trigger das Pluszeichen (+) aus, und wählen Sie dann "Aktion hinzufügen" aus.
Wählen Sie im Bereich "Aktion hinzufügen " die HTTP-Aktion aus. Sollten Sie es nicht sehen, können Sie danach suchen.
Wählen Sie im resultierenden Bereich HTTP aus.

Konfigurieren Sie Ihre Aktion mit den folgenden Einstellungen:

Einstellung Wert

URI Der Endpunkt der veröffentlichten Pipeline. Siehe Voraussetzungen.

Methode POST

Authentifizierungstyp (unter erweiterten Einstellungen) Verwaltete Identität

Einstellung	Wert
URI	Der Endpunkt der veröffentlichten Pipeline. Siehe Voraussetzungen.
Methode	POST
Authentifizierungstyp (unter erweiterten Einstellungen)	Verwaltete Identität

Konfigurieren Sie Ihren Zeitplan, um die Werte aller DataPath PipelineParameters festzulegen, die Sie haben:

{
  "DataPathAssignments": {
    "input_datapath": {
      "DataStoreName": "<datastore-name>",
      "RelativePath": "@{triggerBody()?['Name']}" 
    }
  },
  "ExperimentName": "MyRestPipeline",
  "ParameterAssignments": {
    "input_string": "sample_string3"
  },
  "RunSource": "SDK"
}

Verwenden Sie das DataStoreName, das Sie Ihrem Arbeitsbereich hinzugefügt haben, als Voraussetzung.

Screenshot der HTTP-Einstellungen.

Wählen Sie Speichern aus.

Wichtig

Wenn Sie die rollenbasierte Azure-Zugriffssteuerung (Azure RBAC) verwenden, um den Zugriff auf Ihre Pipeline zu verwalten, legen Sie die Berechtigungen für Ihr Pipelineszenario (Schulung oder Bewertung) fest.

Aufrufen von Machine Learning-Pipelines über Azure Data Factory

In einer Azure Data Factory-Pipeline führt die Aktivität Machine Learning-Pipelineausführung eine Azure Machine Learning-Pipeline aus. Diese Aktivität finden Sie auf der Seite zur Erstellung von Azure Data Factory unter Machine Learning im Menü:

Nächste Schritte

In diesem Artikel haben Sie das Azure Machine Learning SDK für Python verwendet, um eine Pipeline auf zwei verschiedene Arten zu planen. Ein Zeitplan wird basierend auf der verstrichenen Uhrzeit ausgelöst. Der andere Zeitplan wird ausgelöst, wenn eine Datei in einem angegebenen Datastore oder in einem Verzeichnis in diesem Speicher geändert wird. Sie haben erfahren, wie Sie das Portal zum Untersuchen von Pipelines und einzelnen Aufträgen verwenden können. Sie haben gelernt, wie Sie einen Plan deaktivieren, damit die Pipeline nicht mehr ausgeführt wird. Schließlich haben Sie eine Azure-Logik-App erstellt, um eine Pipeline auszulösen.

In diesen Artikeln finden Sie weitere Informationen:

Verwenden von Azure Machine Learning-Pipelines für die Batchbewertung
Weitere Informationen zu Pipelines
Erfahren Sie mehr über das Erkunden von Azure Machine Learning mit Jupyter