Informações gerais

Neste laboratório, você vai:

• implementar um pipeline com ramificações;
• filtrar os dados antes da gravação;
• processar dados como um objeto <Row>;
• adicionar parâmetros de linha de comando personalizados a um pipeline.

Pré-requisitos

Ter noções básicas sobre o Java

No laboratório anterior, você criou um pipeline básico sequencial de extração, transformação e carga, e usou um modelo equivalente do Dataflow para ingerir o armazenamento de dados em lote no Google Cloud Storage. Esse pipeline se baseia em uma sequência de transformações:

No entanto, muitos pipelines não vão exibir uma estrutura tão simples. Neste laboratório, você vai criar um pipeline mais sofisticado e não sequencial.

O caso de uso aqui é otimizar o consumo de recursos. Os produtos variam de acordo com a maneira que consomem recursos. Além disso, nem todos os dados são usados do mesmo modo em uma empresa. Algumas informações serão consultadas regularmente, por exemplo, em cargas de trabalho analíticas, e outras serão usadas apenas para recuperação.

Neste laboratório, você vai otimizar o pipeline do primeiro laboratório para o consumo de recursos ao armazenar apenas dados que os analistas usarão no BigQuery e, ao mesmo tempo, arquivar outros dados em um serviço de armazenamento altamente durável e de baixo custo, o Coldline Storage do Google Cloud Storage.

Configuração e requisitos

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:

{project-number}-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.

Configuração do ambiente de desenvolvimento integrado

Neste laboratório, você vai usar principalmente a versão da Web do ambiente de desenvolvimento integrado Theia. Ele é hospedado no Google Compute Engine e contém o repositório do laboratório pré-clonado. Além disso, o Theia oferece suporte de servidor à linguagem Java e um terminal para acesso programático às APIs do Google Cloud com a ferramenta de linha de comando gcloud, similar ao Cloud Shell.

Para acessar o ambiente de desenvolvimento integrado Theia, copie e cole o link exibido no Qwiklabs em uma nova guia.

OBSERVAÇÃO: mesmo depois que o URL aparecer, talvez você precise esperar de três a cinco minutos para o ambiente ser totalmente provisionado. Até isso acontecer, uma mensagem de erro será exibida no navegador.

O repositório do laboratório foi clonado para seu ambiente. Cada laboratório é dividido em uma pasta labs com códigos que você vai concluir e uma pasta solution com um exemplo totalmente funcional para consulta, caso você enfrente dificuldades. Clique no botão File explorer para conferir:

Também é possível criar vários terminais nesse ambiente, como você faria com o Cloud Shell:

Outra opção de visualização é executar gcloud auth list no terminal em que você fez login com uma conta de serviço fornecida. Ela tem as mesmas permissões que a sua conta de usuário do laboratório:

Se em algum momento o ambiente parar de funcionar, tente redefinir a VM que hospeda o ambiente de desenvolvimento integrado no Console do GCE, conforme este exemplo:

Várias transformações processam a mesma PCollection

Neste laboratório, você vai criar um pipeline com ramificações que grava dados no Google Cloud Storage e no BigQuery. Uma maneira de criar um pipeline com ramificações é aplicar duas transformações diferentes à mesma PCollection, resultando em duas PCollections diferentes.

[PCollection1] = [Initial Input PCollection].apply([A Transform]) [PCollection2] = [Initial Input PCollection].apply([A Different Transform])

Como implementar um pipeline com ramificações

Se você tiver dúvidas nesta seção ou nas próximas, consulte a solução (em inglês).

Tarefa 1: adicionar uma ramificação para fazer gravações no Cloud Storage

Para concluir esta tarefa, modifique um pipeline atual adicionando uma ramificação que faça gravações no Cloud Storage.

Abrir o laboratório apropriado

• Crie outro terminal no ambiente de desenvolvimento integrado, caso ainda não tenha feito isso, e copie e cole este comando:

# Mude o diretório para o laboratório cd 2_Branching_Pipelines/labs # Faça o download das dependências mvn clean dependency:resolve export BASE_DIR=$(pwd)

Configurar o ambiente de dados

# Crie buckets do GCS e um conjunto de dados do BQ cd $BASE_DIR/../.. source create_batch_sinks.sh # Gere um evendo dataflow source generate_batch_events.sh # Mude para o diretório que contém a versão de prática do código cd $BASE_DIR

Clique em Verificar meu progresso para ver o objetivo. Configurar o ambiente de dados

1. Abra MyPipeline.java no seu ambiente de desenvolvimento integrado, que pode ser encontrado em 2_Branching_Pipelines/labs/src/main/java/com/mypackage/pipeline.
2. Role para baixo até o método "run()", em que o corpo do pipeline está definido. Atualmente, o formato se parece com este:

pipeline.apply("ReadFromGCS", TextIO.read().from(input)) .apply("ParseJson", ParDo.of(new JsonToCommonLog())) .apply("WriteToBQ", BigQueryIO.<CommonLog>write().to(output).useBeamSchema() .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE) .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

3. Modifique esse código adicionando uma nova transformação de ramificação que grava no Cloud Storage. Para isso, use TextIO.write() antes de cada elemento ser convertido do json para <CommonLog>.

Se tiver dúvidas nesta seção ou nas próximas, consulte a solução no arquivo de treinamento do analista de dados

Primeiro, por que usar esquemas?

Os esquemas fornecem um sistema de tipos para os registros do Beam, seja lá qual for o tipo específico de linguagem de programação. É possível que haja várias classes Java com o mesmo esquema (por exemplo, uma classe de buffer de protocolo ou uma classe POJO), e o Beam permite a conversão entre esses tipos sem dificuldades. Os esquemas também oferecem uma maneira simples de entender os tipos em diferentes APIs de linguagem de programação.

Uma PCollection com um esquema não precisa ter um codificador especificado, já que o Beam sabe como codificar e decodificar linhas do esquema. O Beam usa um codificador especial para codificar tipos de esquema. Antes do surgimento da API Schema, o Beam teria que saber como codificar todos os objetos no pipeline.

Tarefa 2: filtrar dados por campo

No momento, o novo pipeline não consome menos recursos, já que todos os dados são armazenados duas vezes. Para começar a otimizar o consumo de recursos, precisamos reduzir a quantidade de informações duplicadas. O bucket do Google Cloud Storage funciona como um recurso de arquivamento e backup. Por isso, é importante que todos os dados sejam armazenados nele. No entanto, nem todas as informações precisam ser enviadas ao BigQuery.

Vamos supor que as pessoas responsáveis pela análise de dados geralmente verificam quais recursos os usuários acessam no site e como esses padrões de acesso diferem em função da região geográfica e do tempo. Apenas um subconjunto dos campos seria necessário.

É possível gravar uma DoFn que transforma cada objeto e retorna apenas um subconjunto dos campos. No entanto, o Apache Beam fornece uma grande variedade de transformações relacionais para PCollections que tenham um esquema. O fato de cada registro ser composto por campos nomeados permite agregações simples e legíveis que fazem referência aos campos por nome, semelhante às agregações em uma expressão SQL.

Select e DropFields são duas destas transformações:

PCollection<MyClass> pCollection = ...; pCollection.apply("SelectUserState", Select.fieldNames("state"))); PCollection<MyClass> pCollection = ...; pCollection.apply("DropPII", DropFields.fields("ssn", "state"));

Importante: cada um desses exemplos vai retornar o elemento PCollection<Row> em vez dePCollection<MyClass>. A classe Row pode suportar qualquer esquema e ser considerada um objeto esquematizado genérico. Qualquer PCollection com um esquema pode ser transmitida como uma PCollection de linhas.

As duas transformações acima não vão retornar um objeto CommonLog completo, já que os campos estão sendo removidos. Além disso, a transformação vai reverter para retornar uma Row. É possível criar outro esquema nomeado ou registrar um esquema POJO intermediário, mas é mais fácil só usar a Row por enquanto.

• Para concluir a tarefa, adicione as importações a seguir e altere o conjunto de campos salvos no BigQuery. Dessa maneira, apenas as análises a serem usadas vão ser enviadas ao adicionar uma destas transformações ao pipeline:

import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.schemas.transforms.Select; import org.apache.beam.sdk.schemas.transforms.DropFields;

Importante: se você já tiver o método BigQueryIO.<CommonLog> write() encadeado, será necessário alterá-lo para <Row> devido ao novo tipo.

Tarefa 3: filtrar dados por elemento

Há muitas maneiras de filtrar no Apache Beam. A tarefa anterior demonstrou um método: o uso de uma transformação de esquema. Nesta implementação, você filtrou partes de cada elemento, e isso resultou em um novo objeto Row com um esquema e um subconjunto dos campos restantes. Também é possível usá-los para filtrar elementos inteiros, como no exemplo abaixo:

purchases.apply(Filter.<MyObject>create() .whereFieldName(“costCents”, (Long c) -> c > 100 * 20) .whereFieldName(“shippingAddress.country”, (String c) -> c.equals(“de”)); Observação: esta transformação "Filter", "org.apache.beam.sdk.schemas.transforms.Filter", não deve ser confundida com a função "Filter" antiga e sem esquema, "org.apache.beam.sdk.transforms.Filter"

• Para concluir esta tarefa, primeiro adicione as instruções de importação abaixo ao seu código e, em seguida, adicione uma transformação "Filter" ao pipeline. É possível filtrar usando o critério que preferir. Pode ser necessário adicionar dicas de tipo à função lambda, por exemplo: "(Integer c) -> c > 100".

import org.apache.beam.sdk.schemas.transforms.Filter;

Tarefa 4: adicionar parâmetros personalizados de linha de comando

No momento, o pipeline tem vários parâmetros codificados, incluindo o caminho para a entrada e o local da tabela no BigQuery. No entanto, o pipeline seria mais útil se pudesse ler qualquer arquivo json no Cloud Storage. A adição desse recurso exige a implementação de um conjunto de parâmetros de linha de comando.

No momento, o pipeline usa o elemento PipelineOptionsFactory para gerar uma instância de uma classe personalizada chamada "Options", que não altera nada da classe PipelineOptions. Dessa forma, ela é uma instância de PipelineOptions:

public interface Options extends PipelineOptions { } public static void main(String[] args) { Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); run(options); }

A classe PipelineOptions interpreta argumentos de linha de comando que seguem este formato:

--<option>=<value>

No entanto, é uma classe com um pequeno conjunto de parâmetros predefinidos. Consulte as funções "get-" nesta página (em inglês).

Para adicionar um parâmetro personalizado, é necessário realizar duas tarefas.

1. Primeiro, adicione uma variável de estado para a classe "Options", como no exemplo abaixo:

public interface Options extends PipelineOptions { @Description("My custom command line argument.") @Default.String("DEFAULT") String getMyCustomOption(); void setMyCustomOption(String myCustomOption); }

2. Em seguida, registre sua interface com o elemento PipelineOptionsFactory no método "main()" e transmita a interface ao criar o objeto PipelineOptions. Quando você faz isso com PipelineOptionsFactory, o elemento "--help" consegue encontrar e adicionar a interface de opções personalizadas à saída do comando "--help". No PipelineOptionsFactory, a compatibilidade das opções personalizadas com todas as outras opções registradas também é validada.

No código a seguir, você tem um exemplo de como registrar a interface de opções personalizadas com o elemento PipelineOptionsFactory:

PipelineOptionsFactory.register(Options.class); Options options = PipelineOptionsFactory.fromArgs(args) .withValidation() .as(Options.class);

3. Para acessar um parâmetro de linha de comando no código, basta chamar a função "get" do parâmetro:

String myCustomOption = option.getMyCustomOption();

4. Para concluir esta tarefa, primeiro adicione as instruções de importação a seguir. Após isso, adicione parâmetros de linha de comando para o caminho de entrada, para o caminho de saída do Google Cloud Storage e para o nome da tabela do BigQuery. Por fim, atualize o código do pipeline para acessar esses parâmetros em vez de constantes.

import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description;

Tarefa 5: adicionar campos NULLABLE ao pipeline

Você deve ter notado que a tabela do BigQuery criada no último laboratório tinha um esquema com todos os campos REQUIRED como este:

Talvez seja melhor criar um esquema do Apache Beam com campos NULLABLE em que os dados estejam ausentes, tanto para a execução do pipeline quanto para uma tabela resultante do BigQuery com um esquema que reflita isso.

As anotações "javax" podem ser adicionadas à definição da sua classe, que será incorporada ao esquema do Apache Beam da seguinte maneira:

@DefaultSchema(JavaFieldSchema.class) class MyClass { int field1; @javax.annotation.Nullable String field2; }

• Para concluir esta tarefa, marque os campos lat e lon como anuláveis na definição da classe.

Tarefa 6: executar o pipeline na linha de comando

• Para concluir esta tarefa, execute o pipeline na linha de comando e transmita os parâmetros adequados. Além disso, anote o esquema resultante do BigQuery para campos NULLABLE. O código vai ficar assim:

# Configure as variáveis do ambiente export PROJECT_ID=$(gcloud config get-value project) export REGION='us-central1' export BUCKET=gs://${PROJECT_ID} export COLDLINE_BUCKET=${BUCKET}-coldline export PIPELINE_FOLDER=${BUCKET} export MAIN_CLASS_NAME=com.mypackage.pipeline.MyPipeline export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export OUTPUT_PATH=${PIPELINE_FOLDER}-coldline export TABLE_NAME=${PROJECT_ID}:logs.logs_filtered cd $BASE_DIR mvn compile exec:java \ -Dexec.mainClass=${MAIN_CLASS_NAME} \ -Dexec.cleanupDaemonThreads=false \ -Dexec.args=" \ --project=${PROJECT_ID} \ --region=${REGION} \ --stagingLocation=${PIPELINE_FOLDER}/staging \ --tempLocation=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --inputPath=${INPUT_PATH} \ --outputPath=${OUTPUT_PATH} \ --tableName=${TABLE_NAME}" Importante: se o pipeline for criado sem problemas, mas houver muitos erros devido ao código ou à configuração incorreta no serviço do Dataflow, defina o RUNNER como "DirectRunner" para executá-lo localmente e receber feedback mais rápido. Nesse caso, essa abordagem funciona porque o conjunto de dados é pequeno e você não está usando recursos que não são suportados pelo DirectRunner.

Tarefa 7: verificar os resultados do pipeline

1. Acesse a página Jobs do Cloud Dataflow e veja o job enquanto ele está em execução. O gráfico será semelhante a este:

2. Clique no nó que representa a função Filter, que na imagem acima é chamada de FilterFn. Ao olhar o painel do lado direito, você vai perceber que mais elementos foram adicionados como entradas do que gravados como saídas.

3. Agora clique no nó que representa a gravação no Cloud Storage. Como todos os elementos foram gravados, esse número precisa estar de acordo com o número de elementos na entrada para a função "Filter".

4. Quando o processo do pipeline terminar, examine os resultados no BigQuery consultando sua tabela. O número de registros na tabela precisa estar de acordo com o número de elementos gerados pela função de filtro.

Clique em Verificar meu progresso para conferir o objetivo. Executar o pipeline na linha de comando

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.