Ingerir dados CSV para o BigQuery usando o Cloud Data Fusion: ingestão em lote

1. Introdução

Última atualização:28/02/2020

Este codelab demonstra um padrão de ingestão de dados para ingerir em massa dados de saúde formatados em CSV no BigQuery. Vamos usar o pipeline de dados em lote do Cloud Data Fusion neste laboratório. Dados de teste de saúde realistas foram gerados e disponibilizados para você no bucket do Cloud Storage (gs://hcls_testing_data_fhir_10_patients/csv/).

Neste codelab, você vai aprender a:

• Como ingerir dados CSV (carregamento programado em lote) do GCS para o BigQuery usando o Cloud Data Fusion.
• Como criar visualmente um pipeline de integração de dados no Cloud Data Fusion para carregar, transformar e mascarar dados de saúde em massa.

O que você precisa para executar este codelab?

• Você precisa ter acesso a um projeto do GCP.
• Você precisa ter o papel de proprietário do projeto do GCP.
• Dados de saúde em formato CSV, incluindo o cabeçalho.

Se você não tiver um projeto do GCP, siga estas etapas para criar um.

Os dados de saúde em formato CSV foram pré-carregados no bucket do GCS em gs://hcls_testing_data_fhir_10_patients/csv/. Cada arquivo CSV de recurso tem uma estrutura de esquema exclusiva. Por exemplo, "Patients.csv" tem um esquema diferente de "Providers.csv". Os arquivos de esquema pré-carregados podem ser encontrados em gs://hcls_testing_data_fhir_10_patients/csv_schemas.

Se você precisar de um novo conjunto de dados, gere um usando o SyntheaTM. Em seguida, faça upload para o GCS em vez de copiar do bucket na etapa "Copiar dados de entrada".

Referência:

Operações básicas no Cloud Data Fusion (guia de início rápido) usando o Console do Google Cloud Platform.

2. Configuração do projeto do GCP

Inicialize as variáveis de shell para seu ambiente.

Para encontrar o PROJECT_ID, consulte Como identificar projetos.

<!-- CODELAB: Initialize shell variables ->
<!-- Your current GCP Project ID ->
export PROJECT_ID=<PROJECT_ID>
<!-- A new GCS Bucket in your current Project  - INPUT ->
export BUCKET_NAME=<BUCKET_NAME>
<!-- A new BQ Dataset ID - OUTPUT ->
export DATASET_ID=<DATASET_ID>

Crie um bucket do GCS para armazenar dados de entrada e registros de erros usando a ferramenta gsutil.

gsutil mb -l us gs://$BUCKET_NAME

Acessar o conjunto de dados sintéticos.

1. Envie um e-mail para [e-mail] com o endereço de e-mail que você usa para fazer login no console do Cloud e peça para participar.
2. Você vai receber um e-mail com instruções sobre como confirmar a ação.
3. Use a opção para responder ao e-mail e participar do grupo. NÃO clique no botão.
4. Depois de receber o e-mail de confirmação, siga para a próxima etapa do codelab.

Copie os dados de entrada.

gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME

Crie um conjunto de dados do BigQuery.

bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID

Lembrete:você sempre pode gerar um novo conjunto de dados de teste realistas usando o SyntheaTM para este laboratório. Para sua conveniência, fornecemos um conjunto de dados no GCS: gs://hcls_testing_data_fhir_10_patients/csv

3. Configuração do ambiente do Cloud Data Fusion

Siga estas etapas para ativar a API Cloud Data Fusion e conceder as permissões necessárias:

Ative as APIs.

1. Acesse a Biblioteca de APIs do Console do GCP.
2. Selecione um projeto na lista.
3. Na Biblioteca de APIs, selecione a API que você quer ativar. Se precisar de ajuda para encontrar a API, use o campo de pesquisa e/ou os filtros.
4. Na página da API, clique em ATIVAR.

Crie uma instância do Cloud Data Fusion.

1. No console do GCP, selecione seu ProjectID.
2. Selecione "Data Fusion" no menu à esquerda e clique no botão "CRIAR UMA INSTÂNCIA" no meio da página (primeira criação) ou no botão "CRIAR INSTÂNCIA" no menu superior (criação adicional).

3. Informe o nome da instância. Selecione Enterprise.

4. Clique no botão "CRIAR".

Observação:o processo de criação da instância pode levar até 30 minutos.

Configure as permissões da instância.

Depois de criar uma instância, siga estas etapas para conceder à conta de serviço associada a ela permissões no seu projeto:

Observação:a concessão de permissão precisa ser feita apenas uma vez em um projeto.

1. Clique no nome da instância para acessar a página de detalhes dela.

2. Copie a conta de serviço.

3. Acesse a página do IAM do projeto.
4. Na página de permissões do IAM, vamos adicionar a conta de serviço como um novo membro e conceder a ela o papel Agente de serviço da API Data Fusion do Cloud. Clique no botão Adicionar, cole a "conta de serviço" no campo "Novos membros" e selecione "Gerenciamento de serviços" -> "Papel de agente do servidor da API Data Fusion".
5. 
6. Clique em Salvar.

Depois que essas etapas forem concluídas, o Cloud Data Fusion já estará pronto para ser usado. Basta clicar no link Visualizar instância na página de instâncias do Cloud Data Fusion ou na página de detalhes de uma instância.

Configure a regra de firewall.

1. Navegue até o Console do GCP -> Rede VPC -> Regras de firewall para verificar se a regra default-allow-ssh existe ou não.

2. Caso contrário, adicione uma regra de firewall que permita todo o tráfego SSH de entrada para a rede padrão.

Usando a linha de comando:

gcloud beta compute --project={PROJECT_ID} firewall-rules create default-allow-ssh --direction=INGRESS --priority=1000 --network=default --action=ALLOW --rules=tcp:22 --source-ranges=0.0.0.0/0 --enable-logging

Usando a interface: clique em "Criar regra de firewall" e preencha as informações:

4. Criar um esquema para transformação

Agora que temos o ambiente do Cloud Fusion no GCP, vamos criar um esquema. Precisamos desse esquema para transformar os dados CSV.

1. Na janela do Cloud Data Fusion, clique no link "Visualizar instância" na coluna "Ação". Você será redirecionado para outra página. Clique no URL fornecido para abrir a instância do Cloud Data Fusion. Sua escolha de clicar no botão "Iniciar tour" ou "Não, obrigado" no pop-up de boas-vindas.
2. Abra o menu "hambúrguer" e selecione Pipeline -> Studio.

3. Na seção "Transformar" da paleta de plug-ins à esquerda, clique duas vezes no nó do Wrangler, que vai aparecer na interface do Data Pipelines.

4. Passe o cursor sobre o nó do Wrangler e clique em Propriedades. Clique no botão Organizar e selecione um arquivo de origem .csv (por exemplo, patients.csv), que precisa ter todos os campos de dados para criar o esquema desejado.
5. Clique na seta para baixo (transformações de coluna) ao lado de cada nome de coluna (por exemplo, "body").
6. Por padrão, a importação inicial vai presumir que há apenas uma coluna no arquivo de dados. Para analisar como CSV, escolha Analisar → CSV, selecione o delimitador e marque a caixa "Definir a primeira linha como cabeçalho", conforme necessário. Clique no botão "Aplicar".
7. Clique na seta para baixo ao lado do campo "Corpo" e selecione "Excluir coluna" para remover o campo. Além disso, você pode testar outras transformações, como remover colunas, mudar o tipo de dados de algumas colunas (o padrão é "string"), dividir colunas, definir nomes de colunas etc.

8. As guias "Colunas" e "Etapas de transformação" mostram o esquema de saída e a receita do Wrangler. Clique em Aplicar no canto superior direito. Clique no botão "Validar". A mensagem verde "Nenhum erro encontrado" indica sucesso.

9. Em "Propriedades do Wrangler", clique no menu suspenso Ações para Exportar o esquema desejado para o armazenamento local e Importar no futuro, se necessário.
10. Salve a receita do Wrangler para uso futuro.

parse-as-csv :body ',' true
drop body

11. Para fechar a janela de propriedades do Wrangler, clique no botão X.

5. Criar nós para o pipeline

Nesta seção, vamos criar os componentes do pipeline.

1. Na interface do Data Pipelines, no canto superior esquerdo, verifique se Pipeline de dados - Lote está selecionado como o tipo de pipeline.

2. Há diferentes seções no painel à esquerda, como "Filtro", "Origem", "Transformação", "Análise", "Coletor", "Condições e ações", "Gerenciadores de erros" e "Alertas", em que é possível selecionar um ou mais nós para o pipeline.

Nó de origem

1. Selecione o nó de origem.
2. Na seção "Origem" da paleta de plug-ins à esquerda, clique duas vezes no nó Google Cloud Storage, que aparece na interface do Data Pipelines.
3. Aponte para o nó de origem do GCS e clique em Propriedades.

4. Preencha os campos obrigatórios. Defina os seguintes campos:

• Rótulo = {qualquer texto}
• Nome de referência = {qualquer texto}
• ID do projeto = detecção automática
• Caminho = URL do GCS para o bucket no projeto atual. Por exemplo, gs://$BUCKET_NAME/csv/
• Formatar = texto
• Campo de caminho = nome do arquivo
• Path Filename Only = true
• Read Files Recursively = true

5. Clique no botão + para adicionar o campo "filename" ao esquema de saída do GCS.
6. Clique em Documentação para uma explicação detalhada. Clique no botão "Validar". A mensagem verde "Nenhum erro encontrado" indica sucesso.
7. Para fechar as propriedades do GCS, clique no botão X.

Transformar nó

1. Selecione o nó de transformação.
2. Na seção "Transformar" da paleta de plug-ins à esquerda, clique duas vezes no nó Wrangler, que aparece na interface do usuário do Data Pipelines. Conecte o nó de origem do GCS ao nó de transformação do Wrangler.
3. Passe o cursor sobre o nó do Wrangler e clique em Propriedades.
4. Clique no menu suspenso Ações e selecione Importar para importar um esquema salvo (por exemplo: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json) e cole a receita salva da seção anterior.
5. Ou reutilize o nó Wrangler da seção Criar um esquema para transformação.
6. Preencha os campos obrigatórios. Defina os seguintes campos:

• Label = {any text}
• Nome do campo de entrada = {*}
• Pré-condição = {filename != "patients.csv"} para distinguir cada arquivo de entrada (por exemplo, patients.csv, providers.csv, allergies.csv etc.) do nó de origem.

7. Adicione um nó JavaScript para executar o JavaScript fornecido pelo usuário que transforma ainda mais os registros. Neste codelab, usamos o nó JavaScript para receber um carimbo de data/hora de cada atualização de registro. Conecte o nó de transformação do Wrangler ao nó de transformação do JavaScript. Abra as Propriedades do JavaScript e adicione a seguinte função:

function transform(input, emitter, context) {
  input.TIMESTAMP = (new Date()).getTime()*1000;
  emitter.emit(input);
}

8. Adicione o campo TIMESTAMP ao esquema de saída (se ele não existir) clicando no sinal +. Selecione o carimbo de data/hora como o tipo de dados.

9. Clique em Documentação para uma explicação detalhada. Clique no botão "Validar" para validar todas as informações inseridas. A mensagem verde "Nenhum erro encontrado" indica sucesso.
10. Para fechar a janela "Propriedades da transformação", clique no botão X.

Mascaramento e desidentificação de dados

1. Para selecionar colunas de dados individuais, clique na seta para baixo na coluna e aplique regras de mascaramento em "Mascarar seleção de dados" de acordo com seus requisitos (por exemplo, coluna "CPF").

2. Você pode adicionar mais diretivas na janela Receita do nó Wrangler. Por exemplo, use a diretiva "hash" com o algoritmo de hash seguindo esta sintaxe para fins de desidentificação:

hash <column> <algorithm> <encode>

<column>: name of the column
<algorithm>: Hashing algorithm (i.e. MD5, SHA-1, etc.)
<encode>: default is true (hashed digest is encoded as hex with left-padding zeros). To disable hex encoding, set <encode> to false.

Observação: "True" significa analisar a coluna "Corpo" com a opção "Definir a primeira linha como cabeçalho" marcada.

Nó de coletor

1. Selecione o nó de coletor.
2. Na seção "Coletor" da paleta de plug-ins à esquerda, clique duas vezes no nó do BigQuery, que vai aparecer na interface do Data Pipeline.
3. Aponte para o nó de coletor do BigQuery e clique em "Propriedades".

4. Preencha os campos obrigatórios. Defina os seguintes campos:

• Rótulo = {qualquer texto}
• Nome de referência = {qualquer texto}
• ID do projeto = detecção automática
• Dataset = conjunto de dados do BigQuery usado no projeto atual (ou seja, DATASET_ID)
• Tabela = {table name}

5. Clique em Documentação para uma explicação detalhada. Clique no botão "Validar" para validar todas as informações inseridas. A mensagem verde "Nenhum erro encontrado" indica sucesso.

6. Para fechar as propriedades do BigQuery, clique no botão X.

6. Criar um pipeline de dados em lote

Conectar todos os nós em um pipeline

1. Arraste uma seta de conexão > da extremidade direita do nó de origem e solte na extremidade esquerda do nó de destino.
2. Um pipeline pode ter várias ramificações que recebem arquivos de entrada do mesmo nó de origem do GCS.

Observação:crie uma ramificação do nó de origem no pipeline para cada arquivo .csv de entrada.

3. Nomeie o pipeline.

Observação:durante todo esse processo, o esquema é propagado automaticamente conforme você cria o pipeline. Você também pode importar o esquema desejado.

É isso. Você acabou de criar seu primeiro pipeline de dados em lote e pode implantá-lo e executá-lo.

Enviar alertas de pipeline por e-mail (opcional)

Para usar o recurso Pipeline Alert SendEmail, a configuração exige que um servidor de e-mail seja configurado para enviar e-mails de uma instância de máquina virtual. Consulte o link de referência abaixo para mais informações:

Como enviar e-mails usando uma instância | Documentação do Compute Engine

Neste codelab, vamos configurar um serviço de redirecionamento de e-mail pelo Mailgun seguindo estas etapas:

1. Siga as instruções em Como enviar e-mails com o Mailgun | Documentação do Compute Engine para configurar uma conta com o Mailgun e configurar o serviço de redirecionamento de e-mail. Confira outras modificações abaixo.
2. Adicione todos os endereços de e-mail dos destinatários à lista autorizada do Mailgun. Essa lista pode ser encontrada em Mailgun> Sending> Overview, no painel à esquerda.

Depois que os destinatários clicarem em "Concordo" no e-mail enviado de [e-mail], os endereços de e-mail deles serão salvos na lista autorizada para receber e-mails de alerta do pipeline.

3. Etapa 3 da seção "Antes de começar": crie uma regra de firewall da seguinte maneira:

4. Etapa 3 de "Como configurar o Mailgun como um redirecionamento de e-mail com o Postfix". Selecione Site da Internet ou Internet com smarthost, em vez de Somente local, conforme mencionado nas instruções.

5. Etapa 4 de "Como configurar o Mailgun como um redirecionamento de e-mail com o Postfix". Edite vi /etc/postfix/main.cf para adicionar 10.128.0.0/9 ao final de mynetworks.

6. Edite vi /etc/postfix/master.cf para mudar o smtp padrão (25) para a porta 587.

7. No canto superior direito do Data Fusion Studio, clique em Configurar. Clique em Alerta de pipeline e no botão + para abrir a janela Alertas. Selecione SendEmail.

8. Preencha o formulário de configuração de e-mail. Selecione conclusão, sucesso ou falha no menu suspenso Condição de execução para cada tipo de alerta. Se Incluir token de fluxo de trabalho = false, apenas as informações do campo "Mensagem" serão enviadas. Se Include Workflow Token = true, as informações do campo "Mensagem" e as informações detalhadas do token de fluxo de trabalho serão enviadas. Use minúsculas para Protocolo. Use qualquer e-mail falso que não seja o endereço de e-mail da sua empresa em Remetente.

7. Configurar, implantar, executar/programar o pipeline

1. No canto superior direito do estúdio do Data Fusion, clique em Configurar. Selecione "Spark" em "Configuração do mecanismo". Clique em "Salvar" na janela "Configurar".

2. Clique em Visualizar para ver os dados e clique em Visualizar de novo para voltar à janela anterior. Também é possível **executar** o pipeline no modo de visualização.

3. Clique em Registros para ver os registros.
4. Clique em Salvar para salvar todas as mudanças.
5. Clique em Importar para importar a configuração salva do pipeline ao criar um novo.
6. Clique em Exportar para exportar uma configuração de pipeline.
7. Clique em Implantar para implantar o pipeline.
8. Depois da implantação, clique em Executar e aguarde a conclusão do pipeline.

Observação:quando você executa um pipeline, o Cloud Data Fusion provisiona um cluster temporário do Cloud Dataproc, executa o pipeline e desmonta o cluster. Isso pode levar alguns minutos. Durante esse período, você pode observar o status da transição do pipeline de Provisionamento para Iniciando, de Em execução para Desprovisionamento e, por fim, para Concluído.

9. Para duplicar o pipeline, selecione "Duplicar" no botão Ações.
10. Para exportar a configuração do pipeline, selecione "Exportar" no botão Ações.
11. Clique em Gatilhos de entrada ou Gatilhos de saída na borda esquerda ou direita da janela do Studio para definir gatilhos de pipeline, se quiser.
12. Clique em Programar para agendar a execução do pipeline e carregar dados periodicamente.

13. O Resumo mostra gráficos do histórico de execução, registros, registros de erros e avisos.

8. Validação

1. O pipeline de validação foi executado.

2. Valide se o conjunto de dados do BigQuery tem todas as tabelas.

bq ls $PROJECT_ID:$DATASET_ID

     tableId       Type    Labels   Time Partitioning
----------------- ------- -------- -------------------
 Allergies         TABLE
 Careplans         TABLE
 Conditions        TABLE
 Encounters        TABLE
 Imaging_Studies   TABLE
 Immunizations     TABLE
 Medications       TABLE
 Observations      TABLE
 Organizations     TABLE
 Patients          TABLE
 Procedures        TABLE
 Providers         TABLE

3. Receber e-mails de alerta (se configurado).

Como ver os resultados

Para ver os resultados depois da execução do pipeline:

1. Consulte a tabela na interface do BigQuery. ACESSAR A IU DO BIGQUERY
2. Atualize a consulta abaixo com o nome do seu projeto, conjunto de dados e tabela.

9. Limpar

Para evitar que os recursos usados nesse tutorial sejam cobrados na sua conta do Google Cloud Platform:

Depois de concluir o tutorial, limpe os recursos criados no GCP para que eles não consumam sua cota e você não seja cobrado por eles no futuro. As próximas seções descrevem como excluir ou desativar esses recursos.

Excluir o conjunto de dados do BigQuery

Siga estas instruções para excluir o conjunto de dados do BigQuery criado como parte deste tutorial.

Excluir o bucket do GCS

Siga estas instruções para excluir o bucket do GCS criado como parte deste tutorial.

Como excluir a instância do Cloud Data Fusion

Siga estas instruções para excluir a instância do Cloud Data Fusion.

Excluir o projeto

O jeito mais fácil de evitar cobranças é excluindo o projeto que você criou para este tutorial.

Para excluir o projeto:

Cuidado: excluir um projeto tem os seguintes efeitos:

• Tudo no projeto é excluído. Se você tiver usado um projeto atual para seguir este tutorial, a exclusão dele incluirá quaisquer outras ações feitas no projeto.
• Os IDs do projeto personalizados são perdidos. Ao criar o projeto, você pode ter criado um ID do projeto personalizado para ser usado no futuro. Para preservar os URLs que usam o ID do projeto, como um URL appspot.com, exclua recursos específicos do projeto, em vez de excluir o projeto inteiro.

Se você planeja explorar vários tutoriais e guias de início rápido, a reutilização de projetos pode evitar que você exceda os limites da cota do projeto.

1. No Console do GCP, acesse a página Projetos. ACESSAR A PÁGINA "PROJETOS"
2. Na lista de projetos, selecione um e clique em Excluir.
3. Na caixa de diálogo, digite o ID do projeto e clique em Encerrar para excluí-lo.

10. Parabéns

Parabéns! Você concluiu o codelab para ingerir dados de saúde no BigQuery usando o Cloud Data Fusion.

Você importou dados CSV do Google Cloud Storage para o BigQuery.

Você criou visualmente o pipeline de integração de dados para carregar, transformar e mascarar dados de saúde em massa.

Agora você conhece as principais etapas necessárias para começar sua jornada de análise de dados de saúde com o BigQuery no Google Cloud Platform.