Spouštění kanálů strojového učení

Článek
09/01/2024

V tomto článku se dozvíte, jak programově naplánovat spuštění kanálu v Azure. Plán můžete vytvořit na základě uplynulého času nebo změn systému souborů. Plány založené na čase je možné použít k péči o rutinní úlohy, jako je monitorování posunu dat. Plány založené na změnách se dají použít k reakci na nepravidelné nebo nepředvídatelné změny, například k nahrání nových dat nebo úpravám starých dat. Až se naučíte vytvářet plány, dozvíte se, jak je načíst a deaktivovat. Nakonec se dozvíte, jak ke spouštění kanálů používat další služby Azure, Azure Logic App a Azure Data Factory. Aplikace logiky Azure umožňuje složitější logiku nebo chování. Kanály Azure Data Factory umožňují volat kanál strojového učení jako součást většího kanálu orchestrace dat.

Požadavky

Předplatné Azure. Pokud nemáte předplatné Azure, vytvořte si bezplatný účet.
Prostředí Pythonu, ve kterém je nainstalovaná sada Azure Machine Learning SDK pro Python. Další informace najdete v tématu Vytváření a správa opakovaně použitelných prostředí pro trénování a nasazení pomocí služby Azure Machine Learning.
Pracovní prostor Machine Learning s publikovaným kanálem. Pomocí sady Azure Machine Learning SDK můžete použít kanály pro vytváření a spouštění kanálů strojového učení.

Aktivace kanálů pomocí sady Azure Machine Learning SDK pro Python

K naplánování kanálu budete potřebovat odkaz na váš pracovní prostor, identifikátor publikovaného kanálu a název experimentu, ve kterém chcete vytvořit plán. Tyto hodnoty můžete získat pomocí následujícího kódu:

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"

Vytvoření plánu

Pokud chcete kanál spouštět opakovaně, vytvoříte plán. Přidruží Schedule kanál, experiment a trigger. Aktivační událost může být buď popisemScheduleRecurrence čekání mezi úlohami, nebo cestou úložiště dat, která určuje adresář, který má sledovat změny. V obou případech budete potřebovat identifikátor kanálu a název experimentu, ve kterém chcete vytvořit plán.

V horní části souboru Pythonu naimportujte a ScheduleRecurrence naimportujte tyto Schedule třídy:


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

Vytvoření časového plánu

Konstruktor ScheduleRecurrence má povinný frequency argument, který musí být jedním z následujících řetězců: "Minute", "Hour", "Day", "Week" nebo "Month". Vyžaduje také celočíselnou interval hodnotu určující, kolik frequency jednotek by mělo uplynulo mezi zahájením plánu. Volitelné argumenty umožňují konkrétnější informace o počátečních časech, jak je podrobně popsáno v dokumentaci k sadě ScheduleRecurrence SDK.

Vytvořte Schedule úlohu, která zahájí úlohu každých 15 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)

Vytvoření plánu založeného na změnách

Kanály aktivované změnami souborů můžou být efektivnější než plány založené na čase. Pokud chcete udělat něco před změnou souboru nebo když se do datového adresáře přidá nový soubor, můžete ho předzpracovat. Můžete monitorovat jakékoli změny úložiště dat nebo změny v určitém adresáři v úložišti dat. Pokud monitorujete konkrétní adresář, změny v podadresářích tohoto adresáře neaktivují úlohu.

Poznámka:

Plány založené na změnách podporují pouze monitorování služby Azure Blob Storage.

Chcete-li vytvořit soubor-reaktivní Schedule, musíte nastavit datastore parametr ve volání Schedule.create. Pokud chcete monitorovat složku, nastavte path_on_datastore argument.

Argument polling_interval umožňuje zadat v minutách frekvenci, s jakou je úložiště dat kontrolováno změny.

Pokud byl kanál vytvořen pomocí dataPath PipelineParameter, můžete tuto proměnnou nastavit na název změněného souboru nastavením argumentu data_path_parameter_name .

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

Volitelné argumenty při vytváření plánu

Kromě dříve probíraných argumentů můžete argument nastavit status tak, aby "Disabled" vytvořil neaktivní plán. continue_on_step_failure Nakonec můžete předat logickou hodnotu, která přepíše výchozí chování kanálu při selhání.

Zobrazení plánovaných kanálů

Ve webovém prohlížeči přejděte do služby Azure Machine Learning. V části Koncové body navigačního panelu zvolte Koncové body kanálu. Tím přejdete na seznam kanálů publikovaných v pracovním prostoru.

Stránka Pipelines (Kanály) AML

Na této stránce můžete zobrazit souhrnné informace o všech kanálech v pracovním prostoru: názvy, popisy, stav atd. Přejděte k podrobnostem kliknutím na kanál. Na výsledné stránce najdete další podrobnosti o kanálu a můžete přejít k podrobnostem o jednotlivých úlohách.

Deaktivace kanálu

Pokud máte publikovanou Pipeline , ale neplánovanou verzi, můžete ji zakázat takto:

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

Pokud je kanál naplánovaný, musíte nejprve zrušit plán. Načtěte identifikátor plánu z portálu nebo spuštěním následujícího příkazu:

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

Jakmile budete mít schedule_id možnost zakázat, spusťte:

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)

Pokud pak znovu spustíte Schedule.list(ws) , měli byste získat prázdný seznam.

Použití Azure Logic Apps pro složité triggery

Složitější pravidla triggeru nebo chování je možné vytvořit pomocí aplikace logiky Azure.

Pokud chcete k aktivaci kanálu Machine Learning použít aplikaci logiky Azure, budete potřebovat koncový bod REST pro publikovaný kanál Machine Learning. Vytvořte a publikujte kanál. Pak pomocí ID kanálu vyhledejte koncový bod REST vašeho PublishedPipeline účtu:

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

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

Vytvoření aplikace logiky v Azure

Teď vytvořte instanci aplikace logiky Azure. Po zřízení aplikace logiky pomocí následujícího postupu nakonfigurujte aktivační událost pro váš kanál:

Vytvořte spravovanou identitu přiřazenou systémem, která aplikaci poskytne přístup k vašemu pracovnímu prostoru Azure Machine Learning.
Přejděte do zobrazení Návrháře aplikace logiky a vyberte prázdnou šablonu aplikace logiky.
V Návrháři vyhledejte objekt blob. Vyberte trigger Při přidání nebo změně objektu blob (pouze vlastnosti) a přidejte tento trigger do aplikace logiky.
Vyplňte informace o připojení pro účet úložiště objektů blob, který chcete monitorovat pro přidání nebo úpravy objektů blob. Vyberte kontejner, který chcete monitorovat.

Zvolte interval a četnost dotazování na aktualizace, které vám pomůžou.

Poznámka:

Tato aktivační událost bude monitorovat vybraný kontejner, ale nebude monitorovat podsložky.
Přidejte akci HTTP, která se spustí při zjištění nového nebo upraveného objektu blob. Vyberte + Nový krok a pak vyhledejte a vyberte akci HTTP.

Ke konfiguraci akce použijte následující nastavení:

Nastavení	Hodnota
Akce HTTP	POST
Identifikátor URI	koncový bod publikovaného kanálu, který jste našli jako předpoklad
Režim ověřování	Spravovaná identita

Nastavte plán tak, aby nastavil hodnotu libovolného datapath pipelineParameters, které můžete mít:

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

Jako předpoklad použijte přidaný DataStoreNamepracovní prostor.

Nastavení HTTP

Vyberte Uložit a plán je teď připravený.

Důležité

Pokud ke správě přístupu ke svému kanálu používáte řízení přístupu na základě role v Azure (Azure RBAC), nastavte oprávnění pro váš scénář kanálu (trénování nebo bodování).

Volání kanálů strojového učení z kanálů Azure Data Factory

V kanálu služby Azure Data Factory aktivita spuštění kanálu machine learning spouští kanál služby Azure Machine Learning. Tuto aktivitu najdete na stránce vytváření služby Data Factory v kategorii Machine Learning :

Snímek obrazovky znázorňující aktivitu kanálu ML v prostředí pro vytváření ve službě Azure Data Factory

Další kroky

V tomto článku jste použili sadu Azure Machine Learning SDK pro Python k naplánování kanálu dvěma různými způsoby. Jeden plán se opakuje na základě uplynulého času. Ostatní úlohy plánu, pokud je soubor změněn v zadaném Datastore nebo v adresáři v tomto úložišti. Viděli jste, jak pomocí portálu prozkoumat kanál a jednotlivé úlohy. Dozvěděli jste se, jak zakázat plán, aby kanál přestal běžet. Nakonec jste vytvořili aplikaci logiky Azure, která aktivuje kanál.

Další informace naleznete v tématu:

Použití kanálů Azure Machine Learning pro dávkové vyhodnocování

Další informace o kanálech
Další informace o prozkoumání služby Azure Machine Learning pomocí Jupyteru

Sdílet prostřednictvím