Publikowanie potoku¶
W poprzednim ćwiczeniu powstał potok (ang. pipeline). Teraz wdrożysz go za punktem końcowym wsadowym (ang. batch endpoint), dzięki czemu będzie można go uruchamiać na żądanie albo według harmonogramu, bez otwierania notatnika.
Połączenie z obszarem roboczym¶
Na początek połącz się z obszarem roboczym przy użyciu Azure ML SDK v2.
Uwaga: Jeśli od poprzedniego ćwiczenia wygasła uwierzytelniona sesja z subskrypcją Azure, pojawi się prośba o ponowne zalogowanie.
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
ml_client = MLClient.from_config(credential=credential)
print(f"Gotowe do pracy z obszarem roboczym {ml_client.workspace_name}")
Odszukanie zadania potoku do wdrożenia¶
Żeby potok dało się uruchamiać poza notatnikiem, wdraża się jego zadanie (ang. job) za punktem końcowym wsadowym. Azure ML sam zamienia takie zadanie w komponent potoku wielokrotnego użytku, a punkt końcowy (ang. endpoint) daje stały adres REST, pod który można uderzyć z dowolnej aplikacji, żeby rozpocząć kolejny przebieg.
Warunek jest jeden: potok musi być wcześniej choć raz uruchomiony. Stało się to w poprzednim ćwiczeniu, więc wystarczy teraz odnaleźć tamto zadanie.
# Pobierz najnowsze zadanie z eksperymentu z potokiem
experiment_name = "diabetes-training-pipeline"
pipeline_jobs = [job for job in ml_client.jobs.list() if job.experiment_name == experiment_name]
pipeline_jobs.sort(key=lambda job: job.creation_context.created_at, reverse=True)
pipeline_job_run = ml_client.jobs.get(pipeline_jobs[0].name)
print(f"Użyte zadanie potoku: {pipeline_job_run.name} (status: {pipeline_job_run.status})")
Utworzenie punktu końcowego wsadowego¶
Punkt końcowy wsadowy udostępnia stały adres REST, który nie zmienia się nawet wtedy, gdy stojący za nim potok zostanie później podmieniony albo wdrożony na nowo. Po utworzeniu zobaczysz go na stronie Endpoints (karta Batch endpoints) w Azure Machine Learning studio.
from azure.ai.ml.entities import BatchEndpoint
endpoint_name = "diabetes-pipeline-endpoint"
endpoint = BatchEndpoint(
name=endpoint_name,
description="Batch endpoint for the diabetes training pipeline",
)
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
print(f"Utworzono punkt końcowy '{endpoint_name}'.")
Adres, pod który wywołuje się punkt końcowy, jest zwykłą właściwością jego obiektu:
endpoint = ml_client.batch_endpoints.get(name=endpoint_name)
print(endpoint.scoring_uri)
Wdrożenie potoku w punkcie końcowym¶
Teraz utwórz w punkcie końcowym wdrożenie na podstawie istniejącego zadania potoku. Azure ML potraktuje odnalezione wcześniej zadanie jako definicję komponentu potoku do uruchomienia - nie trzeba więc ponownie definiować komponentów ani funkcji z dekoratorem @dsl.pipeline.
from azure.ai.ml.entities import PipelineComponentBatchDeployment
deployment_name = "diabetes-pipeline-deployment"
deployment = PipelineComponentBatchDeployment(
name=deployment_name,
description="Deployment of the diabetes training pipeline",
endpoint_name=endpoint_name,
job_definition=pipeline_job_run,
settings={"continue_on_step_failure": False, "default_compute": "aml-cluster"},
)
ml_client.batch_deployments.begin_create_or_update(deployment).result()
# Ustaw to wdrożenie jako domyślne dla punktu końcowego
endpoint = ml_client.batch_endpoints.get(name=endpoint_name)
endpoint.defaults.deployment_name = deployment_name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
print(f"Wdrożenie '{deployment_name}' jest teraz domyślne dla punktu końcowego '{endpoint_name}'.")
Korzystanie z punktu końcowego wsadowego¶
Aplikacja kliencka uruchamia potok, wywołując invoke (albo równoważne wywołanie REST). Utworzony wcześniej MLClient jest już uwierzytelniony poświadczeniami Azure, więc nie trzeba samodzielnie pobierać tokenu ani dokładać nagłówka autoryzacji.
Potok wykonuje się asynchronicznie: w odpowiedzi od razu dostajesz obiekt zadania, przez który możesz śledzić jego przebieg.
from azure.ai.ml import Input
from azure.ai.ml.constants import AssetTypes
# Przekaż lokalny plik CSV wprost jako wejście potoku - SDK wyśle go do chmury
# automatycznie. (Krok trenujący w tym potoku czyta pojedynczy uri_file, więc
# nie korzystamy ze współdzielonego zasobu diabetes_mltable, który jest
# zarejestrowany jako mltable.)
job = ml_client.batch_endpoints.invoke(
endpoint_name=endpoint_name,
inputs={"pipeline_input_data": Input(type=AssetTypes.URI_FILE, path="data/diabetes.csv")},
)
print(job.name)
Skoro masz obiekt zadania, możesz wyświetlać na bieżąco jego logi i obserwować przebieg potoku.
Uwaga: Potok powinien zakończyć się szybko, o ile kroki pozwalają na ponowne wykorzystanie wcześniejszych wyników. W praktyce krok trenujący zwykle chcemy uruchamiać za każdym razem, bo dane mogły się zmienić.
ml_client.jobs.stream(name=job.name)
Harmonogram uruchomień potoku¶
Załóżmy, że przychodnia opiekująca się pacjentami z cukrzycą co tydzień zbiera nowe dane i dopisuje je do zbioru. Model warto wtedy trenować ponownie co tydzień - wystarczy dodać harmonogram (ang. schedule), który automatycznie wyzwoli kolejne uruchomienie zadania potoku.
from azure.ai.ml.entities import JobSchedule, RecurrenceTrigger, RecurrencePattern
from azure.ai.ml.constants import TimeZone
schedule_name = "weekly-diabetes-training"
# Wyzwalaj w każdy poniedziałek o 00:00 UTC
recurrence_trigger = RecurrenceTrigger(
frequency="week",
interval=1,
schedule=RecurrencePattern(hours=0, minutes=0, week_days=["Monday"]),
time_zone=TimeZone.UTC,
)
job_schedule = JobSchedule(
name=schedule_name,
trigger=recurrence_trigger,
create_job=pipeline_job_run.name,
)
job_schedule = ml_client.schedules.begin_create_or_update(schedule=job_schedule).result()
print(f"Utworzono harmonogram '{job_schedule.name}'.")
Harmonogramy zdefiniowane w obszarze roboczym wypiszesz w ten sposób:
for schedule in ml_client.schedules.list():
print(schedule.name)
Harmonogram mógł już wyzwolić uruchomienie, więc sprawdź najnowsze zadanie w eksperymencie z potokiem.
recent_jobs = [job for job in ml_client.jobs.list() if job.experiment_name == experiment_name]
recent_jobs.sort(key=lambda job: job.creation_context.created_at, reverse=True)
latest_job = recent_jobs[0]
print(f"Nazwa: {latest_job.name}, status: {latest_job.status}, nazwa wyświetlana: {latest_job.display_name}")
Więcej informacji: O planowaniu uruchomień zadań potoków przeczytasz w artykule Schedule machine learning pipeline jobs w dokumentacji.