Utlösa Machine Learning-pipelines

GÄLLER FÖR:Azure Machine Learning SDK v1 för Python

I den här artikeln får du lära dig hur du programmatiskt schemalägger en pipeline som ska köras i Azure. Du kan skapa ett schema baserat på förfluten tid eller filsystemändringar. Du kan använda tidsbaserade scheman för att utföra rutinuppgifter, till exempel övervakning av dataavvikelser. Du kan använda ändringsbaserade scheman för att reagera på oregelbundna eller oförutsägbara ändringar, till exempel nya data som laddas upp eller gamla data som redigeras.

När du har lärt dig hur du skapar scheman får du lära dig hur du hämtar och inaktiverar dem. Slutligen får du lära dig hur du använder andra Azure-tjänster, Azure Logic Apps och Azure Data Factory, för att köra pipelines. En logikapp möjliggör mer komplex utlösande logik eller beteende. Med Azure Data Factory-pipelines kan du anropa en maskininlärningspipeline som en del av en större dataorkestreringspipeline.

Förutsättningar

• En Azure-prenumeration. Om du inte har en Azure-prenumeration skapar du ett kostnadsfritt konto.

• En Python-miljö där Azure Machine Learning SDK för Python är installerat. Mer information finns i Skapa och hantera återanvändbara miljöer för träning och distribution med Azure Machine Learning.

• En Machine Learning-arbetsyta med en publicerad pipeline. Du kan använda den som skapas i Skapa och köra maskininlärningspipelines med Azure Machine Learning SDK.

Hämta nödvändiga värden

För att schemalägga en pipeline behöver du en referens till din arbetsyta, identifieraren för den publicerade pipelinen och namnet på experimentet där du vill skapa schemat. Du kan hämta dessa värden med hjälp av följande kod:

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" 

Skapa en tidsplan

Om du vill köra en pipeline regelbundet skapar du ett schema. En Schedule associerar en pipeline, ett experiment och en utlösare. Utlösaren kan antingen vara en ScheduleRecurrence som definierar väntetiden mellan jobb eller en datalagersökväg som anger en katalog som ska bevaka ändringar. I båda fallen behöver du pipelineidentifieraren och namnet på experimentet där schemat ska skapas.

Längst upp i Python-filen importerar du klasserna Schedule och ScheduleRecurrence :

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

Skapa ett tidsbaserat schema

Konstruktorn ScheduleRecurrence har ett obligatoriskt frequency argument som måste anges till någon av följande strängar: "Minute", "Hour", "Day", "Week", eller "Month". Det kräver också ett heltalsargument interval som anger hur många frequency enheter som ska förflutit mellan starttiderna. Med valfria argument kan du vara mer specifik om starttider, enligt beskrivningen i dokumentationen om ScheduleRecurrence.

Skapa en Schedule som börjar ett jobb var 15:e minut:

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)

Skapa ett ändringsbaserat schema

Pipelines som utlöses av filändringar kan vara effektivare än tidsbaserade scheman. När du vill göra något innan en fil ändras, eller när en ny fil läggs till i en datakatalog, kan du förbearbeta filen. Du kan övervaka ändringar i ett datalager eller ändringar i en specifik katalog i datalagringen. Om du övervakar en specifik katalog utlöser ändringar i underkataloger i katalogen inte ett jobb.

Om du vill skapa en filreaktiv Schedulemåste du ange parametern datastore i anropet till Schedule.create. Om du vill övervaka en mapp anger du path_on_datastore argumentet.

Med polling_interval argumentet kan du i minuter ange hur ofta datalagringen ska kontrolleras efter ändringar.

Om pipelinen har konstruerats med en DataPathPipelineParameter kan du ange variabeln till namnet på den ändrade filen genom att ange data_path_parameter_name argumentet.

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")

Valfria argument för att skapa ett schema

Förutom argumenten som beskrevs tidigare kan du ange status argumentet till "Disabled" för att skapa ett inaktivt schema. continue_on_step_failure möjliggör att du kan skicka ett booleskt värde som åsidosätter pipelinens standardbeteende vid fel.

Visa dina schemalagda pipelines

I en webbläsare går du till Azure Machine Learning Studio. I den vänstra rutan väljer du ikonen Slutpunkter. I fönstret Slutpunkter väljer du Realtidsslutpunkter. Detta tar dig till en lista över pipelines som publiceras på arbetsytan.

På den här sidan kan du se översiktsinformation om alla pipelines på arbetsytan: namn, beskrivningar, status och så vidare. Du kan få mer information genom att välja namnet på en pipeline. På den resulterande sidan kan du också få information om enskilda jobb.

Inaktivera pipelinen

Om du har en Pipeline som har publicerats men inte schemalagts kan du inaktivera den med hjälp av den här koden:

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

Om pipelinen är schemalagd måste du avbryta schemat först. Hämta schemats identifierare från portalen eller genom att köra den här koden:

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

När du har det schedule_id schema som du vill inaktivera kör du den här koden:

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)

Om du sedan kör Schedule.list(ws) igen bör du få en tom lista.

Använda Logic Apps för komplexa utlösare

Du kan skapa mer komplexa utlösarregler eller beteende med hjälp av Logic Apps.

Om du vill använda en logikapp för att utlösa en Machine Learning-pipeline behöver du REST-slutpunkten för en publicerad Machine Learning-pipeline. Skapa och publicera din pipeline. Leta sedan reda på REST-slutpunkten för din PublishedPipeline med hjälp av pipeline-ID:t:

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

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

Skapa en logikapp i Azure

Skapa nu en logikapp. När logikappen har etablerats använder du de här stegen för att konfigurera en utlösare för din pipeline:

1. Skapa en systemtilldelad hanterad identitet för att ge appen åtkomst till din Azure Machine Learning-arbetsyta.

2. I den vänstra rutan väljer du Logikappmallar i avsnittet Utvecklingsverktyg .

3. Välj mallen Tomt arbetsflöde :

   

4. I designern väljer du Lägg till en utlösare.

5. I fönstret Lägg till en utlösare söker du efter blob. Välj utlösaren När en blob läggs till eller ändras (endast egenskaper).

   

6. I fönstret Skapa anslutning anger du anslutningsinformationen för bloblagringskontot som du vill övervaka för blobtillägg eller ändringar och väljer sedan Skapa ny. Välj den container som ska övervakas.

   Välj Intervall- och frekvensvärden som fungerar åt dig.

   Kommentar

   Den här utlösaren övervakar den valda containern men övervakar inte undermappar.

7. Lägg till en HTTP-åtgärd som körs när en blob ändras eller en ny blob identifieras. Välj plustecknet (+) under utlösaren och välj sedan Lägg till en åtgärd.

8. I fönstret Lägg till en åtgärd väljer du HTTP-åtgärden . Du kan söka efter den om du inte ser den.

   

9. I den resulterande rutan väljer du HTTP.

   Använd följande inställningar för att konfigurera åtgärden:

   Inställning	Värde
   URI	Den publicerade pipelinens slutpunkt. Se Förutsättningar.
   Metod	POST
   Autentiseringstyp (under Avancerade inställningar)	Hanterad identitet

10. Konfigurera ditt schema för att ange värdena för alla DataPath PipelineParameters som du har:

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

    Använd DataStoreName som du har lagt till på din arbetsyta som en förutsättning.

    

11. Välj Spara.

Anropa pipelines för maskininlärning från Azure Data Factory-pipelines

I en Azure Data Factory-pipeline kör machine learning pipeline-aktiviteten en Azure Machine Learning-pipeline. Du hittar den här aktiviteten på redigeringssidan för Azure Data Factory under Machine Learning på menyn:

Nästa steg

I den här artikeln använde du Azure Machine Learning SDK för Python för att schemalägga en pipeline på två olika sätt. Ett schema utlöses baserat på förfluten klocktid. Det andra schemat utlöses om en fil ändras på en angiven Datastore eller i en katalog i den lagringen. Du såg hur du använder portalen för att undersöka pipelinen och enskilda jobb. Du har lärt dig hur du inaktiverar ett schema så att pipelinen slutar köras. Slutligen skapade du en Azure Logic App för att utlösa en pipeline.

De här artiklarna innehåller mer information:

• Använda Azure Machine Learning Pipelines för batchbedömning
• Läs mer om pipelines
• Läs mer om att utforska Azure Machine Learning med Jupyter