Partilhar via

Acionar pipelines de aprendizagem automática

  • Artigo

APLICA-SE A: Python SDK azureml v1

Neste artigo, você aprenderá como agendar programaticamente um pipeline para ser executado no Azure. Você pode criar uma agenda com base no tempo decorrido ou em alterações no sistema de arquivos. As agendas baseadas em tempo podem ser usadas para cuidar de tarefas de rotina, como o monitoramento de desvio de dados. As agendas baseadas em alterações podem ser usadas para reagir a alterações irregulares ou imprevisíveis, como novos dados sendo carregados ou dados antigos sendo editados. Depois de aprender a criar horários, você aprenderá como recuperá-los e desativá-los. Finalmente, você aprenderá a usar outros serviços do Azure, o Aplicativo Lógico do Azure e o Azure Data Factory, para executar pipelines. Um Aplicativo Lógico do Azure permite uma lógica ou comportamento de acionamento mais complexo. Os pipelines do Azure Data Factory permitem que você chame um pipeline de aprendizado de máquina como parte de um pipeline de orquestração de dados maior.

Pré-requisitos

Acionar pipelines com o SDK do Azure Machine Learning para Python

Para agendar um pipeline, você precisará de uma referência ao seu espaço de trabalho, o identificador do pipeline publicado e o nome do experimento no qual deseja criar o agendamento. Você pode obter esses valores com o seguinte código:

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"

Criar uma agenda

Para executar um pipeline de forma recorrente, você criará uma agenda. Um Schedule associa um pipeline, um experimento e um gatilho. O gatilho pode ser umScheduleRecurrence que descreve a espera entre trabalhos ou um caminho de armazenamento de dados que especifica um diretório para observar as alterações. Em ambos os casos, você precisará do identificador de pipeline e do nome do experimento no qual criar a agenda.

Na parte superior do seu arquivo Python, importe as Schedule classes e ScheduleRecurrence :


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

Criar uma agenda baseada no tempo

O ScheduleRecurrence construtor tem um argumento necessário frequency que deve ser uma das seguintes cadeias de caracteres: "Minute", "Hour", "Day", "Week" ou "Month". Também requer um argumento inteiro interval especificando quantas das unidades devem decorrer entre os frequency inícios da programação. Os argumentos opcionais permitem que você seja mais específico sobre os horários de início, conforme detalhado nos documentos do SDK do ScheduleRecurrence.

Crie um Schedule que inicia um trabalho a cada 15 minutos:

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)

Criar uma agenda baseada em alterações

Os pipelines acionados por alterações de arquivo podem ser mais eficientes do que agendamentos baseados em tempo. Quando quiser fazer algo antes de um arquivo ser alterado ou quando um novo arquivo for adicionado a um diretório de dados, você poderá pré-processar esse arquivo. Você pode monitorar quaisquer alterações em um armazenamento de dados ou alterações dentro de um diretório específico dentro do armazenamento de dados. Se você monitorar um diretório específico, as alterações nos subdiretórios desse diretório não acionarão um trabalho.

Nota

As agendas baseadas em alterações dão suporte apenas ao monitoramento do armazenamento de Blob do Azure.

Para criar um arquivo reativo Schedule, você deve definir o datastore parâmetro na chamada como Schedule.create. Para monitorar uma pasta, defina o path_on_datastore argumento.

O polling_interval argumento permite especificar, em minutos, a frequência com que o armazenamento de dados é verificado quanto a alterações.

Se o pipeline foi construído com um DataPath PipelineParameter, você pode definir essa variável como o nome do arquivo alterado definindo o data_path_parameter_name argumento.

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

Argumentos opcionais ao criar uma agenda

Além dos argumentos discutidos anteriormente, você pode definir o status argumento para "Disabled" criar uma agenda inativa. Finalmente, o continue_on_step_failure permite que você passe um Boolean que substituirá o comportamento de falha padrão do pipeline.

Exibir seus pipelines agendados

No navegador da Web, navegue até Azure Machine Learning. Na seção Pontos de extremidade do painel de navegação, escolha Pontos de extremidade de pipeline. Isso leva você a uma lista dos pipelines publicados no espaço de trabalho.

Página de oleodutos da AML

Nesta página, você pode ver informações resumidas sobre todos os pipelines no espaço de trabalho: nomes, descrições, status e assim por diante. Analise detalhadamente clicando em seu pipeline. Na página resultante, há mais detalhes sobre seu pipeline e você pode detalhar trabalhos individuais.

Desativar o pipeline

Se você tiver um Pipeline que está publicado, mas não agendado, você pode desativá-lo com:

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

Se o pipeline estiver agendado, você deverá cancelá-lo primeiro. Recupere o identificador da agenda do portal ou executando:

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

Depois de ter o schedule_id que deseja desativar, execute:

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)

Se você executar novamente Schedule.list(ws) , você deve obter uma lista vazia.

Utilizar o Azure Logic Apps para acionadores complexos

Regras de gatilho ou comportamento mais complexos podem ser criados usando um Aplicativo Lógico do Azure.

Para usar um Aplicativo Lógico do Azure para disparar um pipeline de Aprendizado de Máquina, você precisará do ponto de extremidade REST para um pipeline de Aprendizado de Máquina publicado. Crie e publique seu pipeline. Em seguida, localize o ponto de extremidade REST do seu PublishedPipeline usando o ID do pipeline:

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

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

Criar um aplicativo lógico no Azure

Agora crie uma instância do Aplicativo Lógico do Azure. Depois que seu aplicativo lógico for provisionado, use estas etapas para configurar um gatilho para seu pipeline:

  1. Crie uma identidade gerenciada atribuída ao sistema para dar ao aplicativo acesso ao seu Espaço de Trabalho do Azure Machine Learning.

  2. Navegue até a visualização Logic App Designer e selecione o modelo Blank Logic App.

    Modelo em branco

  3. No Designer, procure por blob. Selecione o gatilho Quando um blob é adicionado ou modificado (somente propriedades) e adicione esse gatilho ao seu Aplicativo Lógico.

    Adicionar gatilho

  4. Preencha as informações de conexão da conta de armazenamento de Blob que você deseja monitorar para adições ou modificações de blob. Selecione o Contêiner a ser monitorado.

    Escolha o Intervalo e a Frequência para pesquisar atualizações que funcionem para você.

    Nota

    Esse gatilho monitorará o contêiner selecionado, mas não monitorará subpastas.

  5. Adicione uma ação HTTP que será executada quando um blob novo ou modificado for detetado. Selecione + Nova Etapa e, em seguida, procure e selecione a ação HTTP.

Procurar ação HTTP

Use as seguintes configurações para configurar sua ação:

Definição Value
Ação HTTP POST
URI o ponto de extremidade para o pipeline publicado que você encontrou como pré-requisito
Modo de autenticação Identidade Gerida

  1. Configure sua agenda para definir o valor de qualquer DataPath PipelineParameters que você possa ter:

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

    Use o DataStoreName que você adicionou ao seu espaço de trabalho como um pré-requisito.

    Definições de HTTP

  2. Selecione Salvar e sua agenda já está pronta.

Importante

Se você estiver usando o controle de acesso baseado em função do Azure (Azure RBAC) para gerenciar o acesso ao seu pipeline, defina as permissões para seu cenário de pipeline (treinamento ou pontuação).

Chamar pipelines de machine learning a partir de pipelines do Azure Data Factory

Em um pipeline do Azure Data Factory, a atividade do Pipeline de Execução do Aprendizado de Máquina executa um pipeline do Aprendizado de Máquina do Azure. Você pode encontrar essa atividade na página de criação do Data Factory na categoria Machine Learning :

Captura de tela mostrando a atividade do pipeline de ML no ambiente de criação do Azure Data Factory

Próximos passos

Neste artigo, você usou o SDK do Azure Machine Learning para Python para agendar um pipeline de duas maneiras diferentes. Um cronograma é recorrente com base no tempo de relógio decorrido. Os outros trabalhos de agendamento se um arquivo for modificado em um diretório especificado Datastore ou dentro de um diretório nesse armazenamento. Você viu como usar o portal para examinar o pipeline e trabalhos individuais. Você aprendeu como desabilitar um agendamento para que o pipeline pare de ser executado. Finalmente, você criou um Aplicativo Lógico do Azure para disparar um pipeline.

Para obter mais informações, consulte:

Usar os Pipelines do Azure Machine Learning para pontuação em lote