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

Článek
08/24/2023

PLATÍ PRO:Sada Python SDK azureml v1

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 na základě změn systému souborů. Pomocí časových plánů se můžete postarat 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, jako jsou nahrání nových dat nebo úpravy starých dat. Až se naučíte vytvářet plány, naučíte se je načítat a deaktivovat. Nakonec se naučíte používat další služby Azure, Azure Logic App a Azure Data Factory, ke spouštění kanálů. Aplikace logiky Azure umožňuje složitější logiku nebo chování aktivace. Azure Data Factory kanály 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 nasazování pomocí služby Azure Machine Learning.
Pracovní prostor Machine Learning s publikovaným kanálem. Můžete použít ten, který je součástí vytváření a spouštění kanálů strojového učení pomocí sady Sdk služby Azure Machine Learning.

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

K naplánování kanálu budete potřebovat odkaz na 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. Objekt Schedule přidruží kanál, experiment a trigger. Triggerem může býtScheduleRecurrence buď popis čekání mezi úlohami, nebo cesta k úložišti dat, která určuje adresář, který má watch pro 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 Schedule třídy a ScheduleRecurrence :


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číselný interval argument, který určuje, kolik frequency jednotek by mělo uplynou 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.

Schedule Vytvořte úlohu, která spustí ú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 něco udělat před změnou souboru nebo když se do datového adresáře přidá nový soubor, můžete tento soubor předzpracovat. Můžete monitorovat všechny 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 se v úložišti dat kontrolují změny.

Pokud byl kanál vytvořen pomocí parametru 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ě výše popsaných argumentů můžete nastavit status argument na , "Disabled" abyste vytvořili neaktivní plán. Nakonec vám umožní předat logickou hodnotu, continue_on_step_failure která přepíše výchozí chování kanálu při selhání.

Zobrazení naplánovaných kanálů

Ve webovém prohlížeči přejděte na 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) V AML

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

Deaktivace kanálu

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

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

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

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

Jakmile budete mít možnost schedule_id , kterou chcete 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 příkaz spustíte Schedule.list(ws) znovu, měl by se zobrazit 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 svůj kanál. Pak pomocí ID kanálu vyhledejte PublishedPipeline koncový bod REST:

# 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 trigger pro váš kanál:

Vytvořte spravovanou identitu přiřazenou systémem , která aplikaci udělí přístup k vašemu pracovnímu prostoru Služby Azure Machine Learning.
Přejděte do zobrazení Designer aplikace logiky a vyberte šablonu Prázdná aplikace logiky.
V Designer vyhledejte objekt blob. Vyberte trigger Při přidání nebo úpravě objektu blob (pouze vlastnosti) a přidejte tento trigger do aplikace logiky.
Vyplňte informace o připojení k účtu úložiště Blob, který chcete monitorovat z hlediska přidání nebo úprav objektů blob. Vyberte kontejner, který chcete monitorovat.

Zvolte Interval a Četnost dotazování na aktualizace, které vám budou fungovat.

Poznámka

Tento trigger 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 všech parametrů kanálu DataPath , které můžete mít:

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

DataStoreName Jako Předpoklad použijte položku, kterou jste přidali do pracovního prostoru.

Nastavení HTTP

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

Důležité

Pokud ke správě přístupu ke 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 Azure Data Factory aktivita Spuštění kanálu služby Machine Learning spouští kanál služby Azure Machine Learning. Tuto aktivitu najdete na stránce pro vytváření ve službě Data Factory v kategorii Machine Learning :

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

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. Další úlohy plánování, pokud je soubor změněn v zadaném Datastore nebo v adresáři v daném úložišti. Viděli jste, jak pomocí portálu prozkoumat kanál a jednotlivé úlohy. Zjistili jste, jak zakázat plán, aby se kanál zastavil. Nakonec jste vytvořili aplikaci logiky Azure, která aktivuje kanál.

Další informace naleznete v tématu:

Použití kanálů Služby Azure Machine Learning k dávkovému vyhodnocování

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