Obtenha mais controle de suas cargas de trabalho do Amazon SageMaker Data Wrangler com conjuntos de dados parametrizados e trabalhos agendados

Os dados estão transformando todos os campos e todos os negócios. No entanto, com os dados crescendo mais rápido do que a maioria das empresas pode acompanhar, coletar dados e obter valor desses dados é uma tarefa desafiadora. UMA estratégia de dados moderna pode ajudá-lo a criar melhores resultados de negócios com dados. A AWS fornece o conjunto mais completo de serviços para o jornada de dados de ponta a ponta para ajudá-lo desbloqueie o valor de seus dados e transformá-lo em insight.

Os cientistas de dados podem gastar até 80% de seu tempo preparando dados para projetos de aprendizado de máquina (ML). Esse processo de preparação é em grande parte indiferenciado e um trabalho tedioso e pode envolver várias APIs de programação e bibliotecas personalizadas. Gerenciador de dados do Amazon SageMaker ajuda os cientistas e engenheiros de dados a simplificar e acelerar a preparação de dados tabulares e de séries temporais e a engenharia de recursos por meio de uma interface visual. Você pode importar dados de várias fontes de dados, como Serviço de armazenamento simples da Amazon (Amazon S3), Amazona atena, Amazon RedShift, ou mesmo soluções de terceiros como Floco de neve or Blocos de dadose processe seus dados com mais de 300 transformações de dados integradas e uma biblioteca de trechos de código, para que você possa normalizar, transformar e combinar recursos rapidamente sem escrever nenhum código. Você também pode trazer suas transformações personalizadas no PySpark, SQL ou Pandas.

Esta postagem demonstra como você pode agendar seus trabalhos de preparação de dados para serem executados automaticamente. Também exploramos o novo recurso Data Wrangler de conjuntos de dados parametrizados, que permite especificar os arquivos a serem incluídos em um fluxo de dados por meio de URIs parametrizados.

Visão geral da solução

O Data Wrangler agora dá suporte à importação de dados usando um URI parametrizado. Isso permite maior flexibilidade porque agora você pode importar todos os conjuntos de dados correspondentes aos parâmetros especificados, que podem ser do tipo String, Number, Datetime e Pattern, no URI. Além disso, agora você pode acionar seus trabalhos de transformação do Data Wrangler em um agendamento.

Nesta postagem, criamos um fluxo de amostra com o conjunto de dados do Titanic para mostrar como você pode começar a experimentar esses dois novos recursos do Data Wrangler. Para baixar o conjunto de dados, consulte Titanic – Aprendizagem de máquina a partir de desastres.

Pré-requisitos

Para obter todos os recursos descritos neste post, você precisa estar executando a versão mais recente do kernel do Data Wrangler. Para obter mais informações, consulte Atualizar Data Wrangler. Além disso, você precisa estar executando Estúdio Amazon SageMaker JupyterLab 3. Para visualizar a versão atual e atualizá-la, consulte Versão JupyterLab.

Estrutura de arquivo

Para esta demonstração, seguimos uma estrutura de arquivo simples que você deve replicar para reproduzir as etapas descritas neste post.

1. Em estúdio, criar um novo caderno.
2. Execute o seguinte trecho de código para criar a estrutura de pastas que usamos (certifique-se de estar na pasta desejada em sua árvore de arquivos):

   !mkdir titanic_dataset
   !mkdir titanic_dataset/datetime_data
   !mkdir titanic_dataset/datetime_data/2021
   !mkdir titanic_dataset/datetime_data/2022
   
   !mkdir titanic_dataset/datetime_data/2021/01 titanic_dataset/datetime_data/2021/02 titanic_dataset/datetime_data/2021/03 
   !mkdir titanic_dataset/datetime_data/2021/04 titanic_dataset/datetime_data/2021/05 titanic_dataset/datetime_data/2021/06
   !mkdir titanic_dataset/datetime_data/2022/01 titanic_dataset/datetime_data/2022/02 titanic_dataset/datetime_data/2022/03 
   !mkdir titanic_dataset/datetime_data/2022/04 titanic_dataset/datetime_data/2022/05 titanic_dataset/datetime_data/2022/06
   
   !mkdir titanic_dataset/datetime_data/2021/01/01 titanic_dataset/datetime_data/2021/02/01 titanic_dataset/datetime_data/2021/03/01 
   !mkdir titanic_dataset/datetime_data/2021/04/01 titanic_dataset/datetime_data/2021/05/01 titanic_dataset/datetime_data/2021/06/01
   !mkdir titanic_dataset/datetime_data/2022/01/01 titanic_dataset/datetime_data/2022/02/01 titanic_dataset/datetime_data/2022/03/01 
   !mkdir titanic_dataset/datetime_data/2022/04/01 titanic_dataset/datetime_data/2022/05/01 titanic_dataset/datetime_data/2022/06/01
   
   !mkdir titanic_dataset/train_1 titanic_dataset/train_2 titanic_dataset/train_3 titanic_dataset/train_4 titanic_dataset/train_5
   !mkdir titanic_dataset/train titanic_dataset/test

3. Copie o train.csv e test.csv arquivos do conjunto de dados original do Titanic para as pastas titanic_dataset/train e titanic_dataset/test, Respectivamente.
4. Execute o seguinte trecho de código para preencher as pastas com os arquivos necessários:

   import os
   import math
   import pandas as pd
   batch_size = 100
   
   #Get a list of all the leaf nodes in the folder structure
   leaf_nodes = []
   
   for root, dirs, files in os.walk('titanic_dataset'):
       if not dirs:
           if root != "titanic_dataset/test" and root != "titanic_dataset/train":
               leaf_nodes.append(root)
               
   titanic_df = pd.read_csv('titanic_dataset/train/train.csv')
   
   #Create the mini batch files
   for i in range(math.ceil(titanic_df.shape[0]/batch_size)):
       batch_df = titanic_df[i*batch_size:(i+1)*batch_size]
       
       #Place a copy of each mini batch in each one of the leaf folders
       for node in leaf_nodes:
           batch_df.to_csv(node+'/part_{}.csv'.format(i), index=False)

Nós dividimos o train.csv arquivo do conjunto de dados do Titanic em nove arquivos diferentes, chamados part_x, onde x é o número da peça. A parte 0 contém os primeiros 100 registros, a parte 1 os próximos 100 e assim por diante até a parte 8. Cada pasta de nó da árvore de arquivos contém uma cópia das nove partes dos dados de treinamento, exceto para o train e test pastas, que contêm train.csv e test.csv.

Conjuntos de dados parametrizados

Os usuários do Data Wrangler agora podem especificar parâmetros para os conjuntos de dados importados do Amazon S3. Os parâmetros do conjunto de dados são especificados no URI dos recursos, e seu valor pode ser alterado dinamicamente, permitindo maior flexibilidade para selecionar os arquivos que queremos importar. Os parâmetros podem ser de quatro tipos de dados:

• Sessão – Pode assumir o valor de qualquer número inteiro
• Tanga – Pode assumir o valor de qualquer string de texto
• padrão – Pode assumir o valor de qualquer expressão regular
• Data hora – Pode assumir o valor de qualquer um dos formatos de data/hora suportados

Nesta seção, fornecemos um passo a passo desse novo recurso. Isso está disponível somente depois que você importa seu conjunto de dados para seu fluxo atual e apenas para conjuntos de dados importados do Amazon S3.

1. No seu fluxo de dados, escolha o sinal de mais (+) ao lado da etapa de importação e escolha Editar conjunto de dados.
2. O método preferido (e mais fácil) de criar novos parâmetros é destacando uma seção do seu URI e escolhendo Criar parâmetro personalizado no menu suspenso. Você precisa especificar quatro coisas para cada parâmetro que deseja criar:
   1. Nome
   2. Formato
   3. Valor padrão
   4. Descrição

   
   Aqui criamos um parâmetro do tipo String chamado filename_param com um valor padrão de train.csv. Agora você pode ver o nome do parâmetro entre colchetes duplos, substituindo a parte do URI que destacamos anteriormente. Como o valor definido para este parâmetro foi train.csv, agora vemos o arquivo train.csv listados na tabela de importação.
   

3. Quando tentamos criar um trabalho de transformação, no Configurar trabalho passo, agora vemos um parâmetros seção, onde podemos ver uma lista de todos os nossos parâmetros definidos.
4. A escolha do parâmetro nos dá a opção de alterar o valor do parâmetro, neste caso, alterando o conjunto de dados de entrada a ser transformado de acordo com o fluxo definido.
   Supondo que mudemos o valor de filename_param da train.csv para part_0.csv, o trabalho de transformação agora leva part_0.csv (desde que um arquivo com o nome part_0.csv existe na mesma pasta) como seus novos dados de entrada.
   
5. Além disso, se você tentar exportar seu fluxo para um destino do Amazon S3 (por meio de um notebook Jupyter), agora verá uma nova célula contendo os parâmetros que você definiu.
   Observe que o parâmetro leva seu valor padrão, mas você pode alterá-lo substituindo seu valor no parameter_overrides dicionário (deixando as chaves do dicionário inalteradas).
   
   Além disso, você pode criar novos parâmetros a partir do parâmetros UI.
6. Abra-o escolhendo o ícone de parâmetros ({{}}) localizado próximo ao Go opção; ambos estão localizados próximos ao valor do caminho do URI.
   Uma tabela é aberta com todos os parâmetros que existem atualmente em seu arquivo de fluxo (filename_param neste ponto).
7. Você pode criar novos parâmetros para seu fluxo escolhendo Criar Parâmetro.
   
   Uma janela pop-up é aberta para permitir que você crie um novo parâmetro personalizado.
8. Aqui, criamos um novo example_parameter como tipo de número com um valor padrão de 0. Este parâmetro recém-criado agora está listado no parâmetros tabela. Passar o mouse sobre o parâmetro exibe as opções Editar, Apagar e inserção.
9. De dentro do parâmetros UI, você pode inserir um de seus parâmetros no URI selecionando o parâmetro desejado e escolhendo inserção.
   Isso adiciona o parâmetro ao final do seu URI. Você precisa movê-lo para a seção desejada em seu URI.
   
10. Altere o valor padrão do parâmetro, aplique a alteração (do modal), escolha Goe escolha o ícone de atualização para atualizar a lista de visualização usando o conjunto de dados selecionado com base no valor do parâmetro recém-definido.Vamos agora explorar outros tipos de parâmetros. Suponha que agora temos um conjunto de dados dividido em várias partes, onde cada arquivo tem um número de peça.
11. Se quisermos alterar dinamicamente o número do arquivo, podemos definir um parâmetro Number conforme mostrado na captura de tela a seguir.Observe que o arquivo selecionado é aquele que corresponde ao número especificado no parâmetro.
    Agora vamos demonstrar como usar um parâmetro Pattern. Suponha que queremos importar todos os part_1.csv arquivos em todas as pastas sob o titanic-dataset/ pasta. Os parâmetros de padrão podem receber qualquer expressão regular válida; existem alguns padrões regex mostrados como exemplos.
12. Crie um parâmetro padrão chamado any_pattern para corresponder a qualquer pasta ou arquivo sob o titanic-dataset/ pasta com valor padrão .*.Observe que o curinga não é um único * (asterisco), mas também tem um ponto.
13. Destaque o titanic-dataset/ parte do caminho e crie um parâmetro personalizado. Desta vez escolhemos o padrão tipo.Este padrão seleciona todos os arquivos chamados part-1.csv de qualquer uma das pastas em titanic-dataset/.
    Um parâmetro pode ser usado mais de uma vez em um caminho. No exemplo a seguir, usamos nosso parâmetro recém-criado any_pattern duas vezes em nosso URI para corresponder a qualquer um dos arquivos de peça em qualquer uma das pastas em titanic-dataset/.
    Finalmente, vamos criar um parâmetro Datetime. Parâmetros datetime são úteis quando estamos lidando com caminhos particionados por data e hora, como os gerados por Mangueira de incêndio de dados do Amazon Kinesis (Vejo Particionamento dinâmico no Kinesis Data Firehose). Para esta demonstração, usamos os dados na pasta datetime-data.
14. Selecione a parte do seu caminho que é uma data/hora e crie um parâmetro personalizado. Escolha o Data hora tipo de parâmetro.
    Ao escolher o tipo de dados Datetime, você precisa preencher mais detalhes.
15. Primeiro de tudo, você deve fornecer um formato de data. Você pode escolher qualquer um dos formatos de data/hora predefinidos ou criar um personalizado.
    Para os formatos predefinidos de data/hora, a legenda fornece um exemplo de data correspondente ao formato selecionado. Para esta demonstração, escolhemos o formato aaaa/MM/dd.
16. Em seguida, especifique um fuso horário para os valores de data/hora.
    Por exemplo, a data atual pode ser 1º de janeiro de 2022, em um fuso horário, mas pode ser 2 de janeiro de 2022, em outro fuso horário.
17. Por fim, você pode selecionar o intervalo de tempo, que permite selecionar o intervalo de arquivos que deseja incluir em seu fluxo de dados.
    Você pode especificar seu intervalo de tempo em horas, dias, semanas, meses ou anos. Para este exemplo, queremos obter todos os arquivos do ano passado.
18. Forneça uma descrição do parâmetro e escolha Crie.
    Se você estiver usando vários conjuntos de dados com diferentes fusos horários, a hora não será convertida automaticamente; você precisa pré-processar cada arquivo ou fonte para convertê-lo em um fuso horário.Os arquivos selecionados são todos os arquivos nas pastas correspondentes aos dados do ano anterior.
19. Agora, se criarmos um trabalho de transformação de dados, podemos ver uma lista de todos os nossos parâmetros definidos e podemos substituir seus valores padrão para que nossos trabalhos de transformação escolham os arquivos especificados.
    

Agendar tarefas de processamento

Agora você pode agendar trabalhos de processamento para automatizar a execução dos trabalhos de transformação de dados e exportar seus dados transformados para o Amazon S3 ou Loja de recursos Amazon SageMaker. Você pode agendar os trabalhos com o horário e a periodicidade que mais lhe convier.

Trabalhos de processamento agendados usam Amazon Event Bridge regras para agendar a execução do trabalho. Portanto, como pré-requisito, você deve certificar-se de que o Gerenciamento de acesso e identidade da AWS (IAM) sendo usado pelo Data Wrangler, ou seja, o Amazon Sage Maker papel de execução da instância do Studio, tem permissões para criar regras do EventBridge.

Configurar IAM

Prossiga com as seguintes atualizações na função de execução do IAM SageMaker correspondente à instância do Studio em que o fluxo do Data Wrangler está em execução:

1. Anexar o AmazonEventBridgeFullAccess política gerenciada.
2. Anexe uma política para conceder permissão para criar um trabalho de processamento:

   {
   	"Version": "2012-10-17",
   	"Statement": [
   		{
   			"Effect": "Allow",
   			"Action": "sagemaker:StartPipelineExecution",
   			"Resource": "arn:aws:sagemaker:Region:AWS-account-id:pipeline/data-wrangler-*"
   		}
   	]
   }

3. Conceda permissão ao EventBridge para assumir a função adicionando a seguinte política de confiança:

   {
   	"Effect": "Allow",
   	"Principal": {
   		"Service": "events.amazonaws.com"
   	},
   	"Action": "sts:AssumeRole"
   }

Como alternativa, se você estiver usando uma função diferente para executar o trabalho de processamento, aplique as políticas descritas nas etapas 2 e 3 a essa função. Para obter detalhes sobre a configuração do IAM, consulte Crie uma programação para processar novos dados automaticamente.

Crie uma programação

Para criar um planejamento, abra seu fluxo no editor de fluxo Data Wrangler.

1. No Fluxo de dados guia, escolha Criar emprego.
2. Configure os campos obrigatórios e escolha Em seguida, 2. Configurar trabalho.
   
3. Expandir Horários Associados.
4. Escolha Criar nova programação.
   
   A Criar nova programação A caixa de diálogo é aberta, onde você define os detalhes da programação do trabalho de processamento.
   A caixa de diálogo oferece grande flexibilidade para ajudá-lo a definir o cronograma. Você pode ter, por exemplo, o trabalho de processamento sendo executado em um horário específico ou a cada X horas, em dias específicos da semana.
   A periodicidade pode ser granular ao nível de minutos.
5. Defina o nome do cronograma e a periodicidade e escolha Crie para salvar a programação.
6. Você tem a opção de iniciar o trabalho de processamento imediatamente junto com o agendamento, que cuida das execuções futuras, ou deixar o trabalho para ser executado apenas de acordo com o agendamento.
7. Você também pode definir uma programação adicional para o mesmo trabalho de processamento.
   
8. Para concluir a programação do job de processamento, selecione Crie.
   Você vê uma mensagem “Trabalho agendado com sucesso”. Além disso, se você optar por deixar a tarefa para ser executada apenas de acordo com o agendamento, verá um link para a regra EventBridge que acabou de criar.

Se você escolher o link de agendamento, uma nova guia no navegador será aberta, mostrando a regra do EventBridge. Nesta página, você pode fazer outras modificações na regra e acompanhar seu histórico de invocação. Para interromper a execução do trabalho de processamento planejado, exclua a regra de evento que contém o nome do planejamento.

A regra EventBridge mostra um pipeline SageMaker como seu destino, que é acionado de acordo com o agendamento definido e o trabalho de processamento invocado como parte do pipeline.

Para rastrear as execuções do pipeline do SageMaker, você pode voltar ao Studio, escolher o Recursos do SageMaker ícone, escolha Dutose escolha o nome do pipeline que deseja rastrear. Agora você pode ver uma tabela com todas as execuções atuais e passadas e o status desse pipeline.

Você pode ver mais detalhes clicando duas vezes em uma entrada específica.

limpar

Quando você não estiver usando o Data Wrangler, é recomendável desligar a instância na qual ele é executado para evitar incorrer em taxas adicionais.

Para evitar a perda de trabalho, salve seu fluxo de dados antes de encerrar o Data Wrangler.

1. Para salvar seu fluxo de dados no Studio, escolha Envie o, Em seguida, escolha Salvar fluxo do Data Wrangler. O Data Wrangler salva automaticamente seu fluxo de dados a cada 60 segundos.
2. Para encerrar a instância do Data Wrangler, no Studio, escolha Executando instâncias e kernels.
3. Debaixo APLICATIVOS DE CORRIDA, escolha o ícone de desligamento ao lado do sagemaker-data-wrangler-1.0 app.
4. Escolha Desligar tudo para confirmar.

O Data Wrangler é executado em uma instância ml.m5.4xlarge. Esta instância desaparece de INSTÂNCIAS EM EXECUÇÃO quando você desliga o aplicativo Data Wrangler.

Depois de encerrar o aplicativo Data Wrangler, ele precisa ser reiniciado na próxima vez que você abrir um arquivo de fluxo do Data Wrangler. Isso pode levar alguns minutos.

Conclusão

Nesta postagem, demonstramos como você pode usar parâmetros para importar seus conjuntos de dados usando fluxos do Data Wrangler e criar trabalhos de transformação de dados neles. Os conjuntos de dados parametrizados permitem mais flexibilidade nos conjuntos de dados que você usa e permitem que você reutilize seus fluxos. Também demonstramos como você pode configurar trabalhos agendados para automatizar suas transformações e exportações de dados para Amazon S3 ou Feature Store, no horário e na periodicidade adequados às suas necessidades, diretamente da interface do usuário do Data Wrangler.

Para saber mais sobre como usar fluxos de dados com o Data Wrangler, consulte Criar e usar um fluxo do Data Wrangler e Preços do Amazon SageMaker. Para começar a usar o Data Wrangler, consulte Prepare dados de ML com o Amazon SageMaker Data Wrangler.

---

Sobre os autores

David Laredo é arquiteto de prototipagem da equipe de prototipagem e engenharia de nuvem da Amazon Web Services, onde ajudou a desenvolver vários protótipos de aprendizado de máquina para clientes da AWS. Ele trabalha com aprendizado de máquina há 6 anos, treinando e ajustando modelos de ML e implementando pipelines de ponta a ponta para produzir esses modelos. Suas áreas de interesse são NLP, aplicativos de ML e ML de ponta a ponta.

Givanildo Alves é Arquiteto de Prototipagem da equipe de Prototipagem e Engenharia de Nuvem da Amazon Web Services, ajudando clientes a inovar e acelerar mostrando a arte do possível na AWS, tendo já implementado diversos protótipos em torno da inteligência artificial. Ele tem uma longa carreira em engenharia de software e trabalhou anteriormente como Engenheiro de Desenvolvimento de Software na Amazon.com.br.

Adriano Fontes é gerente de programa da equipe de prototipagem e engenharia de nuvem da Amazon Web Services, inovando para clientes em aprendizado de máquina, IoT e blockchain. Ele tem mais de 15 anos de experiência em gerenciamento e implementação de projetos e 1 ano de experiência na AWS.