Visão geral

Os fluxos de trabalho são um tema comum quando falamos em análise de dados. Eles envolvem ingestão, transformação e análise de dados para descobrir informações relevantes. A ferramenta do Google Cloud para hospedagem de fluxos de trabalho é o Cloud Composer. Ele é a versão hospedada da famosa ferramenta de fluxo de trabalho de código aberto Apache Airflow.

Neste laboratório, você vai usar o console do Google Cloud para configurar um ambiente do Cloud Composer. Depois, você vai usar o Cloud Composer para realizar um fluxo de trabalho simples que verifica a existência de um arquivo de dados, cria um cluster do Cloud Dataproc, executa um job de contagem de palavras e depois exclui esse cluster.

Atividades deste laboratório

• Usar o console do Google Cloud para criar o ambiente do Cloud Composer

• Abrir e executar o gráfico acíclico dirigido (DAG) na interface da Web do Airflow

• Acessar os resultados do job de contagem de palavras no armazenamento

Configuração e requisitos

Configuração do laboratório

Para cada laboratório, você recebe um novo projeto do Google Cloud e um conjunto de recursos por um determinado período e sem custos financeiros.

1. Faça login no Qwiklabs em uma janela anônima.

2. Confira o tempo de acesso do laboratório (por exemplo, 1:15:00) e finalize todas as atividades nesse prazo.
   Não é possível pausar o laboratório. Você pode reiniciar o desafio, mas vai precisar refazer todas as etapas.

3. Quando tudo estiver pronto, clique em Começar o laboratório.

4. Anote as credenciais (Nome de usuário e Senha). É com elas que você vai fazer login no Console do Google Cloud.

5. Clique em Abrir Console do Google.

6. Clique em Usar outra conta, depois copie e cole as credenciais deste laboratório nos locais indicados.
   Se você usar outras credenciais, vai receber mensagens de erro ou cobranças.

7. Aceite os termos e pule a página de recursos de recuperação.

Ative o Google Cloud Shell

O Google Cloud Shell é uma máquina virtual com ferramentas de desenvolvimento. Ele tem um diretório principal permanente de 5 GB e é executado no Google Cloud.

O Cloud Shell oferece acesso de linha de comando aos recursos do Google Cloud.

1. No console do Cloud, clique no botão "Abrir o Cloud Shell" na barra de ferramentas superior direita.

   

2. Clique em Continuar.

O provisionamento e a conexão do ambiente podem demorar um pouco. Quando você estiver conectado, já estará autenticado, e o projeto estará definido com seu PROJECT_ID. Exemplo:

A gcloud é a ferramenta de linha de comando do Google Cloud. Ela vem pré-instalada no Cloud Shell e aceita preenchimento com tabulação.

• Para listar o nome da conta ativa, use este comando:

gcloud auth list

Saída:

Credentialed accounts: - @.com (active)

Exemplo de saída:

Credentialed accounts: - google1623327_student@qwiklabs.net

• Para listar o ID do projeto, use este comando:

gcloud config list project

Saída:

[core] project =

Exemplo de saída:

[core] project = qwiklabs-gcp-44776a13dea667a6 Observação: a documentação completa da gcloud está disponível no guia com informações gerais sobre a gcloud CLI .

Verifique as permissões do projeto

Antes de começar a trabalhar no Google Cloud, veja se o projeto tem as permissões corretas no Identity and Access Management (IAM).

1. No console do Google Cloud, em Menu de navegação (), selecione IAM e administrador > IAM.

2. Confira se a conta de serviço padrão do Compute {project-number}-compute@developer.gserviceaccount.com está na lista e recebeu o papel de editor. O prefixo da conta é o número do projeto, que está no Menu de navegação > Visão geral do Cloud > Painel.

Observação: se a conta não estiver no IAM ou não tiver o papel de editor, siga as etapas abaixo.

1. No console do Google Cloud, em Menu de navegação, clique em Visão geral do Cloud > Painel.
2. Copie o número do projeto, por exemplo, 729328892908.
3. Em Menu de navegação, clique em IAM e administrador > IAM.
4. Clique em Permitir acesso, logo abaixo de Visualizar por principais na parte de cima da tabela de papéis.
5. Em Novos principais, digite:

{número-do-projeto}-compute@developer.gserviceaccount.com

6. Substitua {project-number} pelo número do seu projeto.
7. Em Papel, selecione Projeto (ou Básico) > Editor.
8. Clique em Save.

Tarefa 1: atividades do ambiente

1. Na barra de título do console do Google Cloud, clique em Ativar o Cloud Shell. Se for solicitado, clique em Continuar.

2. Execute os comandos a seguir para atribuir o papel Worker do Composer à conta de serviço de desenvolvedor do Compute.

export PROJECT_ID=$(gcloud config get-value project) export PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)") gcloud projects add-iam-policy-binding {{{project_0.project_id}}} \ --member=serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com \ --role=roles/composer.worker

3. Execute os comandos abaixo para garantir que as APIs necessárias sejam ativadas corretamente no projeto.

gcloud services disable composer.googleapis.com gcloud services disable artifactregistry.googleapis.com gcloud services disable container.googleapis.com gcloud services enable artifactregistry.googleapis.com gcloud services enable container.googleapis.com gcloud services enable composer.googleapis.com

Tarefa 2: criar o ambiente do Cloud Composer

Nesta seção, você vai criar um ambiente do Cloud Composer.

Observação: antes de continuar, verifique se você executou as tarefas anteriores para ativar as APIs necessárias. Caso contrário, faça isso para evitar falha na criação do ambiente do Cloud Composer.

1. Na barra de título do console do Google Cloud, digite Composer no campo de pesquisa e clique em Composer na seção "Produtos e página".

2. Clique em Criar ambiente e selecione Composer 3. Defina o seguinte para seu ambiente:

Propriedade	Valor
Nome	highcpu
Local
Versão de imagem	composer-3-airflow-n.n.n-build.n (observação: selecione a imagem com o maior número disponível)

3. Em Recursos do ambiente, escolha Pequeno.

Não altere as outras configurações.

4. Clique em Criar.

O processo de criação do ambiente será concluído quando uma marca de seleção verde for exibida à esquerda do nome dele na página "Ambientes" no console.

O processo de configuração do ambiente pode levar de 15 a 30 minutos. Continue o laboratório enquanto o ambiente é gerado.

Clique em Verificar meu progresso para conferir o objetivo.

Crie o ambiente do Cloud Composer.

Crie um bucket do Cloud Storage

Crie um bucket do Cloud Storage no projeto. Ele será usado como saída no job do Hadoop no Dataproc.

1. Acesse Menu de navegação > Cloud Storage > Buckets e clique em + Criar.

2. Dê um nome exclusivo ao bucket, como o ID do projeto (), e clique em Criar. Se a mensagem O acesso público será bloqueado aparecer, clique em Confirmar.

Não esqueça o nome do bucket do Cloud Storage. Ele será usado posteriormente neste laboratório como uma variável do Airflow.

Clique em Verificar meu progresso para conferir o objetivo.

Crie um bucket do Cloud Storage.

Tarefa 3: definição e conceitos principais do Airflow

Enquanto você aguarda a criação do ambiente do Composer, veja alguns termos usados com o Airflow.

O Airflow é uma plataforma para criar, programar e monitorar fluxos de trabalho de forma programática.

Use o Airflow para criar fluxos de trabalho como gráficos acíclicos dirigidos (DAGs) de tarefas. O programador do Airflow executa suas tarefas em uma matriz de workers enquanto segue as dependências especificadas.

Principais conceitos

DAG

Um gráfico acíclico dirigido é uma coleção das tarefas que você quer executar, organizadas para refletir as próprias relações e dependências.

Operador

É a descrição de uma única tarefa. Em geral, os operadores são atômicos. Por exemplo, o operador BashOperator é usado para executar o comando bash.

Tarefa

Uma instância parametrizada de um operador. É um nó no DAG.

Instância de tarefa

Uma execução específica de uma tarefa, caracterizada como um DAG, uma tarefa e um ponto no tempo. Ela pode ter vários estados, por exemplo: running, success, failed, skipped etc.

Leia a Documentação sobre conceitos (em inglês) para saber mais.

Tarefa 4: definir o fluxo de trabalho

Agora vamos falar sobre o fluxo de trabalho usado. No Cloud Composer, os fluxos de trabalho são compostos de gráficos acíclicos dirigidos (DAGs). Os DAGs são definidos em arquivos Python padrão, armazenados na pasta DAG_FOLDER do Airflow. O Airflow executará o código em cada arquivo para criar dinamicamente os objetos DAG. É possível ter quantos DAGs quiser e usar cada um deles para descrever algumas tarefas. Em geral, cada um deles deve corresponder a um único fluxo de trabalho lógico.

Veja abaixo o código do fluxo de trabalho hadoop_tutorial.py, também chamado de DAG:

"""Exemplo de um DAG do Airflow que cria um cluster do Cloud Dataproc, executa o exemplo de contagem de palavras do Hadoop e exclui o cluster. Este DAG depende de três variáveis do Airflow https://airflow.apache.org/concepts.html#variables * gcp_project - Projeto do Google Cloud que será usado pelo cluster do Cloud Dataproc. * gce_zone - Zona do Google Compute Engine onde o cluster do Cloud Dataproc deve ser criado. * gce_region - Região do Google Compute Engine em que o cluster do Cloud Dataproc será criado. * gcs_bucket - Bucket do Google Cloud Storage utilizado como saída dos jobs do Hadoop a partir do Dataproc. Leia https://cloud.google.com/storage/docs/creating-buckets para aprender a criar um bucket. """ import datetime import os from airflow import models from airflow.contrib.operators import dataproc_operator from airflow.utils import trigger_rule # Arquivo produzido pelo job do Cloud Dataproc. output_file = os.path.join( models.Variable.get('gcs_bucket'), 'wordcount', datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep # Caminho para o exemplo de contagem de palavras do Hadoop disponível em todos os clusters do Dataproc. WORDCOUNT_JAR = ( 'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar' ) # Argumentos que devem ser transmitidos ao job do Cloud Dataproc. wordcount_args = ['wordcount', 'gs://pub/shakespeare/rose.txt', output_file] yesterday = datetime.datetime.combine( datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()) default_dag_args = { # Definir a data de início como ontem inicia o DAG imediatamente quando ele for # detectado no bucket do Cloud Storage. 'start_date': yesterday, # Para enviar e-mail em caso de falha ou nova tentativa, defina o argumento 'email' para seu e-mail e ative # essa opção aqui. 'email_on_failure': False, 'email_on_retry': False, # Se uma tarefa falhar, faça uma nova tentativa depois de esperar pelo menos 5 minutos 'retries': 1, 'retry_delay': datetime.timedelta(minutes=5), 'project_id': models.Variable.get('gcp_project') } with models.DAG( 'composer_hadoop_tutorial', # Continue executando o DAG uma vez por dia schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag: # Crie um cluster do Cloud Dataproc. create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator( task_id='create_dataproc_cluster', # Dê um nome exclusivo ao cluster anexando a data agendada. # Confira https://airflow.apache.org/code.html#default-variables cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', num_workers=2, region=models.Variable.get('gce_region'), zone=models.Variable.get('gce_zone'), image_version='2.0', master_machine_type='e2-standard-2', worker_machine_type='e2-standard-2') # Execute o exemplo de contagem de palavras do Hadoop instalado no nó mestre do cluster do # Cloud Dataproc. run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator( task_id='run_dataproc_hadoop', region=models.Variable.get('gce_region'), main_jar=WORDCOUNT_JAR, cluster_name='composer-hadoop-tutorial- cluster-{{ ds_nodash }}', arguments=wordcount_args) # Exclua o cluster do Cloud Dataproc. delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator( task_id='delete_dataproc_cluster', region=models.Variable.get('gce_region'), cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', # Definir trigger_rule como ALL_DONE faz com que o cluster seja excluído # mesmo que o job do Dataproc apresente falha. trigger_rule=trigger_rule.TriggerRule.ALL_DONE) # Defina as dependências de DAG. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Para orquestrar as três tarefas do fluxo de trabalho, o DAG importa os seguintes operadores:

1. DataprocClusterCreateOperator: cria um cluster do Cloud Dataproc.
2. DataProcHadoopOperator: envia um job de contagem de palavras do Hadoop e grava os resultados em um bucket do Cloud Storage.
3. DataprocClusterDeleteOperator: exclui o cluster para evitar cobranças contínuas do Compute Engine.

As tarefas são executadas sequencialmente, como mostra esta seção do arquivo:

# Defina as dependências do DAG. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

O nome do DAG é composer_hadoop_tutorial, e ele é executado uma vez por dia:

with models.DAG( 'composer_hadoop_tutorial', # Continue executando o DAG uma vez por dia schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag:

Como o código start_date transmitido para default_dag_args está definido como yesterday, o Cloud Composer programa o fluxo de trabalho para iniciar imediatamente após o DAG ser carregado.

Tarefa 5: consultar as informações do ambiente

1. Volte para o Composer e verifique o status do seu ambiente.

2. Depois que seu ambiente for criado, clique no nome dele (highcpu) para ver os detalhes.

Na guia Configuração do ambiente, você encontra informações como o URL da interface da Web do Airflow, o cluster do GKE e um link para a pasta de DAGs, que está armazenada no seu bucket.

Observação: o Cloud Composer só programa os fluxos de trabalho que estão na pasta /dags.

Tarefa 6: usar a interface do Airflow

Para acessar a interface da Web do Airflow pelo console, siga estas etapas:

1. Volte para a página Ambientes.
2. Na coluna Servidor da Web do Airflow do ambiente, clique em Airflow.
3. Clique nas suas credenciais do laboratório.
4. A interface da Web do Airflow será aberta em uma nova janela do navegador.

Tarefa 7: definir variáveis do Airflow

As variáveis do Airflow são um conceito específico da plataforma e diferente das variáveis de ambiente.

1. Na interface do Airflow, selecione Admin > Variables na barra de menus.

2. Clique no ícone + para adicionar um novo registro.

3. Crie as variáveis gcp_project, gcs_bucket, gce_zone e gce_region do Airflow. Clique em Save depois de criar cada variável.

Key	Val	Details
gcp_project		É o projeto do Google Cloud Platform usado neste laboratório.
gcs_bucket	gs://<my-bucket>	Substitua <my-bucket> pelo nome do bucket do Cloud Storage que você criou. Esse bucket armazena a resposta dos jobs do Hadoop no Dataproc.
gce_zone		Essa é a zona do Compute Engine em que o cluster do Cloud Dataproc será criado.
gce_region		Essa é a região do Compute Engine em que o cluster do Cloud Dataproc será criado.

Clique em Save. Depois de adicionar a primeira variável, repita o mesmo processo para a segunda e a terceira. A tabela de variáveis será semelhante a esta no final:

Tarefa 8: fazer upload do DAG para o Cloud Storage

Siga estas etapas para fazer o upload do DAG:

1. No Cloud Shell, execute o comando abaixo para fazer upload de uma cópia do arquivo hadoop_tutorial.py no bucket do Cloud Storage que foi criado automaticamente quando você criou o ambiente.

2. No comando abaixo, substitua <DAGs_folder_path> pelo caminho para a pasta de DAGs:

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py <DAGs_folder_path>

• Você encontra o caminho no Composer.
• Clique no ambiente que você já criou e na guia Configuração do ambiente para ver os detalhes do ambiente.
• Encontre a pasta de DAGs e copie o caminho.

O comando revisado para fazer upload do arquivo é parecido com este:

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py gs://{{{project_0.default_region|REGION}}}-highcpu-0682d8c0-bucket/dags

3. Após o upload para o diretório de DAGs, abra a pasta dags no bucket para ver o arquivo na guia Objetos dos detalhes do bucket.

Quando um arquivo DAG é incluído na pasta relacionada, o Cloud Composer adiciona o DAG ao Airflow e o programa automaticamente. As alterações no DAG levam de 3 a 5 minutos.

É possível conferir o status da tarefa do DAG composer_hadoop_tutorial na interface da Web do Airflow.

Observação: é possível ignorar com segurança qualquer mensagem na interface, como "The scheduler does not appear to be running…". A interface da Web do Airflow é atualizada conforme o DAG avança.

Clique em Verificar meu progresso para conferir o objetivo.

Faça upload do DAG no Cloud Storage.

Como o DAG é executado

Quando estiver na pasta dags no Cloud Storage, o arquivo DAG será analisado pelo Cloud Composer. Se nenhum erro for encontrado, o nome do fluxo de trabalho será exibido na lista de DAGs e entrará na fila para ser executado imediatamente.

1. Verifique se você está na guia "DAGs" na interface da Web do Airflow. É preciso esperar alguns minutos até que esse processo seja concluído. Atualize o navegador para acessar as informações mais recentes.

2. No Airflow, clique em composer_hadoop_tutorial para abrir a página de detalhes do DAG. Ela inclui várias representações das tarefas e dependências do fluxo de trabalho.

3. Na barra de ferramentas, clique em Graph. Passe o cursor sobre a representação gráfica de cada tarefa para conferir o status correspondente. A borda das tarefas também indica o status: verde = em execução, vermelha = falha etc.

4. Clique no link "Refresh" para ver as informações mais recentes. As bordas dos processos mudam de cor para indicar o estado.

Observação: quando o cluster do Dataproc já existe, é possível executar o fluxo de trabalho novamente para alcançar o estado "success". Para isso, clique no gráfico `create_dataproc_cluster` e em Clear, que redefine as três tarefas. Depois, clique em OK para confirmar.

5. Assim que o status de create_dataproc_cluster mudar para "running", acesse o Menu de navegação > Dataproc e clique em:

   • Clusters para monitorar a criação e a exclusão de clusters. O cluster criado pelo fluxo de trabalho é efêmero e será excluído como parte da última tarefa após a execução do fluxo.
   • Jobs para monitorar o job de contagem de palavras do Apache Hadoop. Clique no ID do job para ver a saída do registro dele.

6. Quando o Dataproc alcançar o estado "running", volte para o Airflow e clique em Refresh para conferir se o cluster está completo.

Ao final do processo run_dataproc_hadoop, acesse o Menu de navegação > Cloud Storage > Buckets e clique no nome do seu bucket para conferir os resultados da contagem de palavras na pasta wordcount.

7. Depois que todas as etapas forem concluídas no DAG, cada uma delas terá uma borda verde escura. Além disso, o cluster do Dataproc que foi criado é excluído.

Parabéns!

Você executou um fluxo de trabalho do Cloud Composer.

Próximas etapas

• Confira a apresentação do Cloud Composer no NEXT 18 em São Francisco: Flexible, Easy Data Pipelines on Google Cloud with Cloud Composer (Cloud Next '18).
• Para ver o valor de uma variável, execute o subcomando variables da CLI do Airflow com o argumento "get" ou use a interface da Web do Airflow.
• Para mais informações sobre a interface da Web do Airflow, consulte Como acessar a interface da Web.

Finalize o laboratório

Clique em Terminar o laboratório após a conclusão. O Google Cloud Ensina remove os recursos usados e limpa a conta por você.

Você vai poder avaliar sua experiência no laboratório. Basta selecionar o número de estrelas, digitar um comentário e clicar em Enviar.

O número de estrelas indica o seguinte:

• 1 estrela = muito insatisfeito
• 2 estrelas = insatisfeito
• 3 estrelas = neutro
• 4 estrelas = satisfeito
• 5 estrelas = muito satisfeito

Feche a caixa de diálogo se não quiser enviar feedback.

Para enviar seu feedback, fazer sugestões ou correções, use a guia Suporte.

Copyright 2020 Google LLC. Todos os direitos reservados. Google e o logotipo do Google são marcas registradas da Google LLC. Todos os outros nomes de produtos e empresas podem ser marcas registradas das respectivas empresas a que estão associados.