ETL reverso do Databricks para o Spanner usando CSV

1. Criar um pipeline de ETL reverso do Databricks para o Spanner usando o GCS e o Dataflow

Introdução

Neste codelab, você vai criar um pipeline de ETL reversa do Databricks para o Spanner usando arquivos CSV armazenados no Google Cloud Storage. Tradicionalmente, os pipelines de ETL (extração, transformação e carregamento) movem dados de bancos de dados operacionais para um data warehouse como o Databricks para análise. Um pipeline de ETL reverso faz o contrário: ele move dados tratados e processados do data warehouse de volta para sistemas operacionais, onde podem alimentar aplicativos, veicular recursos voltados ao usuário ou ser usados para tomada de decisões em tempo real.

O objetivo é mover um conjunto de dados de amostra de uma tabela do Databricks para o Spanner, um banco de dados relacional distribuído globalmente ideal para aplicativos de alta disponibilidade.

Para isso, o Google Cloud Storage (GCS) e o Dataflow são usados como etapas intermediárias. Confira um resumo do fluxo de dados e o raciocínio por trás dessa arquitetura:

Databricks para Google Cloud Storage (GCS) em formato CSV:

A primeira etapa é extrair os dados do Databricks em um formato aberto e universal. A exportação para CSV é um método comum e simples para criar arquivos de dados portáteis. Esses arquivos serão armazenados temporariamente no GCS, que oferece uma solução de armazenamento de objetos escalonável e durável.

GCS para Spanner (via Dataflow):

Em vez de escrever um script personalizado para ler do GCS e gravar no Spanner, usamos o Google Dataflow, um serviço de processamento de dados totalmente gerenciado. O Dataflow oferece modelos pré-criados especificamente para esse tipo de tarefa. Usar o modelo "GCS Text to Cloud Spanner" permite uma importação de dados paralela e de alta capacidade sem escrever código de processamento de dados, economizando um tempo de desenvolvimento significativo.

O que você vai aprender

Como carregar dados no Databricks
Como criar um bucket do GCS
Como exportar uma tabela do Databricks para o GCS no formato CSV
Como configurar uma instância do Spanner
Como carregar tabelas CSV no Spanner com o Dataflow

2. Configuração, requisitos e limitações

Pré-requisitos

Uma conta do Databricks com permissões para criar clusters e instalar bibliotecas. Uma conta de teste sem custo financeiro não é suficiente para este laboratório.
Uma conta do Google Cloud com as APIs Spanner, Cloud Storage e Dataflow ativadas.
Acesso ao console do Google Cloud por um navegador da Web.
Um terminal com a Google Cloud CLI instalada.
Se a organização do Google Cloud tiver a política iam.allowedPolicyMemberDomains ativada, um administrador talvez precise conceder uma exceção para permitir contas de serviço de domínios externos. Isso será abordado em uma etapa posterior, quando aplicável.

Permissões do IAM no Google Cloud Platform

A Conta do Google precisa das seguintes permissões para executar todas as etapas deste codelab.

Contas de serviço
`iam.serviceAccountKeys.create`	Permite a criação de contas de serviço.
Spanner
`spanner.instances.create`	Permite criar uma nova instância do Spanner.
`spanner.databases.create`	Permite executar instruções DDL para criar
`spanner.databases.updateDdl`	Permite executar instruções DDL para criar tabelas no banco de dados.
Google Cloud Storage
`storage.buckets.create`	Permite criar um bucket do GCS para armazenar os arquivos Parquet exportados.
`storage.objects.create`	Permite gravar os arquivos Parquet exportados no bucket do GCS.
`storage.objects.get`	Permite que o BigQuery leia os arquivos Parquet do bucket do GCS.
`storage.objects.list`	Permite que o BigQuery liste os arquivos Parquet no bucket do GCS.
Dataflow
`Dataflow.workitems.lease`	Permite reivindicar itens de trabalho do Dataflow.
`Dataflow.workitems.sendMessage`	Permite que o worker do Dataflow envie mensagens de volta para o serviço do Dataflow.
`Logging.logEntries.create`	Permite que os workers do Dataflow gravem entradas de registro no Google Cloud Logging.

Para sua conveniência, é possível usar papéis predefinidos que contêm essas permissões.

roles/resourcemanager.projectIamAdmin

roles/iam.serviceAccountKeyAdmin

roles/spanner.instanceAdmin

roles/spanner.databaseAdmin

roles/storage.admin

roles/dataflow.serviceAgent

roles/dataflow.worker

roles/dataflow.serviceAgent

Limitações

É importante estar ciente das diferenças de tipo de dados ao mover dados entre sistemas.

Databricks para CSV:ao exportar, os tipos de dados do Databricks são convertidos em representações de texto padrão.
CSV para Spanner:ao importar, é necessário garantir que os tipos de dados de destino do Spanner sejam compatíveis com as representações de string no arquivo CSV. Este laboratório mostra um conjunto comum de mapeamentos de tipos.

Configurar propriedades reutilizáveis

Alguns valores serão necessários repetidamente ao longo deste laboratório. Para facilitar, vamos definir esses valores como variáveis de shell para serem usadas mais tarde.

GCP_REGION: a região específica em que os recursos do GCP vão estar localizados. Confira a lista de regiões aqui.
GCP_PROJECT: o ID do projeto do GCP a ser usado.
GCP_BUCKET_NAME: o nome do bucket do GCS a ser criado e onde os arquivos de dados serão armazenados.

export GCP_REGION = <GCP REGION HERE> 
export GCP_PROJECT= <GCP PROJECT HERE>
export GCS_BUCKET_NAME = <GCS BUCKET NAME HERE>
export SPANNER_INSTANCE = <SPANNER INSTANCE ID HERE>
export SPANNER_DB = <SPANNER DATABASE ID HERE>

Databricks

Para este laboratório, uma conta do Databricks hospedada no GCP para permitir a definição de um local de dados externo no GCS.

Google Cloud

Este laboratório requer um projeto do Google Cloud.

Projeto do Google Cloud

Um projeto é uma unidade básica de organização no Google Cloud. Se um administrador tiver fornecido um para uso, essa etapa poderá ser ignorada.

Um projeto pode ser criado usando a CLI desta forma:

gcloud projects create $GCP_PROJECT
gcloud config set project $GCP_PROJECT

Saiba mais sobre como criar e gerenciar projetos aqui.

Configurar o Spanner

Para começar a usar o Spanner, é necessário provisionar uma instância e um banco de dados. Confira detalhes sobre como configurar e criar uma instância do Spanner aqui.

Criar a instância

gcloud spanner instances create $SPANNER_INSTANCE \
--config=regional-$GCP_REGION \
--description="Codelabs Snowflake RETL" \
--processing-units=100 \
--edition=ENTERPRISE

Criar o banco de dados

gcloud spanner databases create $SPANNER_DB \
--instance=$SPANNER_INSTANCE

3. Crie um bucket do Google Cloud Storage

O Google Cloud Storage (GCS) será usado para armazenar temporariamente os arquivos de dados CSV gerados pelo Snowflake antes de serem importados para o Spanner.

Criar o bucket

Use o comando a seguir para criar um bucket de armazenamento em uma região específica.

gcloud storage buckets create gs://$GCS_BUCKET_NAME --location=$GCP_REGION

Verificar a criação do bucket

Depois que o comando for executado, confira o resultado listando todos os buckets. O novo bucket vai aparecer na lista resultante. As referências de bucket geralmente aparecem com o prefixo gs:// antes do nome do bucket.

gcloud storage ls | grep gs://$GCS_BUCKET_NAME

Testar permissões de gravação

Essa etapa garante que o ambiente local esteja autenticado corretamente e tenha as permissões necessárias para gravar arquivos no bucket recém-criado.

echo "Hello, GCS" | gcloud storage cp - gs://$GCS_BUCKET_NAME/hello.txt

Verificar o arquivo enviado

Liste os objetos no bucket. O caminho completo do arquivo que acabou de ser enviado por upload vai aparecer.

gcloud storage ls gs://$GCS_BUCKET_NAME

Você verá esta resposta:

gs://$GCS_BUCKET_NAME/hello.txt

Para ver o conteúdo de um objeto em um bucket, use gcloud storage cat.

gcloud storage cat gs://$GCS_BUCKET_NAME/hello.txt

O conteúdo do arquivo precisa estar visível:

Hello, GCS

Limpar o arquivo de teste

O bucket do Cloud Storage está configurado. Agora é possível excluir o arquivo de teste temporário.

gcloud storage rm gs://$GCS_BUCKET_NAME/hello.txt

A saída precisa confirmar a exclusão:

Removing gs://$GCS_BUCKET_NAME/hello.txt...
/ [1 objects]
Operation completed over 1 objects.

4. Exportar do Databricks para o GCS

Agora, o ambiente do Databricks será configurado para se conectar com segurança ao GCS e exportar dados.

Criar credenciais

No menu à esquerda, clique em Catálogo.
Clique em Dados externos se essa opção estiver disponível na parte de cima da página do catálogo. Caso contrário, clique no menu suspenso Conectar e em Credenciais.
Mude para a guia Credenciais se ela ainda não estiver aberta.
Clique em Criar credenciais.
Selecione GCP Service Account para Tipo de credencial.
Insira codelabs-retl-credentials em Nome da credencial.
Clique em Criar.
Copie o e-mail da conta de serviço da caixa de diálogo e clique em Concluído.

Defina essa conta de serviço como uma variável de ambiente na sua instância de shell para reutilização:

export GCP_SERVICE_ACCOUNT=<Your service account>

Conceder permissões do GCS ao Databricks

Agora, a conta de serviço do Snowflake precisa receber permissão para gravar no bucket do GCS.

gcloud storage buckets add-iam-policy-binding gs://$GCS_BUCKET_NAME \
    --member="serviceAccount:$GCP_SERVICE_ACCOUNT" \
    --role="roles/storage.objectAdmin"

gcloud storage buckets add-iam-policy-binding gs://$GCS_BUCKET_NAME \
    --member="serviceAccount:$GCP_SERVICE_ACCOUNT" \
    --role="roles/storage.legacyBucketReader"

Criar local externo

Volte para a página Credenciais usando o rastreamento de navegação na parte de cima da página.
Mude para a guia Local externo.
Clique em Criar local externo.
Defina o Nome do local externo como codelabs-retl-gcs
Mantenha o Tipo de armazenamento como GCP
Defina o caminho do bucket como o URL.
Defina Credencial de armazenamento como codelabs-retl-credentials.
Clique em Criar.
Na confirmação. Clique em Criar.

Criar catálogo e esquema

No menu à esquerda, clique em Catálogo.
Clique em Criar e em Criar um catálogo.
Defina o Nome do catálogo como retl_tpch_project
Defina Tipo como Standard
Selecione codelabs-retl-gcs como local externo
Clique em Criar.
Clique em retl_tpch_project na lista Catálogo.
Clique em Criar esquema.
Defina o Nome do esquema como tpch_data.
Selecione Local de armazenamento como codelabs-retl-gcs
Clique em Criar.

Exportar dados como CSV

Agora os dados estão prontos para exportação. O conjunto de dados TPC-H de amostra será usado para definir nossa nova tabela, que será armazenada externamente como CSV.

Primeiro, copie os dados de amostra para uma nova tabela no espaço de trabalho. Para isso, o código SQL precisa ser executado de uma consulta.

No menu à esquerda, em SQL, clique em Consultas.
Clique no botão Criar consulta.
Ao lado do botão Executar, defina o Espaço de trabalho como retl_tpch_project.

CREATE TABLE retl_tpch_project.tpch_data.regional_sales_csv
USING CSV
LOCATION 'gs://<Your bucket name>/regional_sales_csv'
OPTIONS (
  header "false",
  delimiter ","
)
AS
SELECT
    n.n_name AS nation_name,
    c.c_mktsegment AS market_segment,
    YEAR(o.o_orderdate) AS order_year,
    o.o_orderpriority AS order_priority,
    COUNT(o.o_orderkey) AS total_order_count,
    ROUND(SUM(o.o_totalprice), 2) AS total_revenue,
    COUNT(DISTINCT c.c_custkey) AS unique_customer_count
FROM samples.tpch.orders AS o
INNER JOIN samples.tpch.customer AS c
    ON o.o_custkey = c.c_custkey
INNER JOIN samples.tpch.nation AS n
    ON c.c_nationkey = n.n_nationkey
GROUP BY 1, 2, 3, 4;

Verificar dados no GCS

Verifique o bucket do GCS para ver os arquivos criados pelo Databricks.

gcloud storage ls gs://$GCS_BUCKET_NAME/regional_sales_csv/

Um ou mais arquivos .csv devem estar visíveis, junto com _SUCCESS e arquivos de registro.

5. Carregar dados no Spanner com o Dataflow

Um modelo do Dataflow fornecido pelo Google será usado para importar os dados CSV do GCS para o Spanner.

Criar a tabela do Spanner

Primeiro, crie a tabela de destino no Spanner. O esquema precisa ser compatível com os dados nos arquivos CSV.

gcloud spanner databases ddl update $SPANNER_DB \
  --instance=$SPANNER_INSTANCE \
  --ddl="$(cat <<EOF
CREATE TABLE regional_sales (
    nation_name STRING(MAX),
    market_segment STRING(MAX),
    order_year INT64,
    order_priority STRING(MAX),
    total_order_count INT64,
    total_revenue NUMERIC,
    unique_customer_count INT64
) PRIMARY KEY (nation_name, market_segment, order_year, order_priority);
EOF
)"

Criar o manifesto do Dataflow

O modelo do Dataflow requer um arquivo de manifesto. É um arquivo JSON que informa ao modelo onde encontrar os arquivos de dados de origem e em qual tabela do Spanner eles devem ser carregados.

Defina e faça upload de um novo regional_sales_manifest.json para o bucket do GCS:

cat <<EOF | gcloud storage cp - gs://$GCS_BUCKET_NAME/regional_sales_manifest.json 
{ 
  "tables": [
    {
       "table_name": "regional_sales", 
       "file_patterns": [ 
         "gs://$GCS_BUCKET_NAME/regional_sales_csv/*.csv"
       ] 
    } 
  ] 
} 
EOF

Ative a API do Dataflow

Antes de usar o Dataflow, é preciso ativá-lo. Faça isso com

gcloud services enable dataflow.googleapis.com --project=$GCP_PROJECT

Criar e executar o job do Dataflow

O job de importação está pronto para ser executado. Esse comando inicia um job do Dataflow usando o modelo GCS_Text_to_Cloud_Spanner.

O comando é longo e tem vários parâmetros. Confira os detalhes:

--gcs-location: o caminho para o modelo pré-criado no GCS.
--region: a região em que o job do Dataflow será executado.
--parameters: uma lista de pares de chave-valor específicos do modelo:
instanceId, databaseId: a instância e o banco de dados de destino do Spanner.
importManifest: o caminho do GCS para o arquivo de manifesto recém-criado.

gcloud dataflow jobs run spanner-import-from-gcs \
  --gcs-location=gs://dataflow-templates/latest/GCS_Text_to_Cloud_Spanner \
  --region=$GCP_REGION \
  --staging-location=gs://$GCS_BUCKET_NAME/staging \
  --parameters \
instanceId=$SPANNER_INSTANCE,\
databaseId=$SPANNER_DB,\
importManifest=gs://$GCS_BUCKET_NAME/regional_sales_manifest.json,escape='\'

O status do job do Dataflow pode ser verificado com o seguinte comando:

gcloud dataflow jobs list \
    --filter="name:spanner-import-from-gcs" \
    --region="$GCP_REGION" \
    --sort-by="~creationTime" \
    --limit=1

O job leva cerca de 5 minutos para ser concluído.

Verificar dados no Spanner

Depois que o job do Dataflow for concluído, verifique se os dados foram carregados no Spanner.

Primeiro, verifique a contagem de linhas. Ela precisa ser 4375.

gcloud spanner databases execute-sql $SPANNER_DB \
--instance=$SPANNER_INSTANCE \
--sql='SELECT COUNT(*) FROM regional_sales;'

Em seguida, consulte algumas linhas para inspecionar os dados.

gcloud spanner databases execute-sql $SPANNER_DB \
--instance=$SPANNER_INSTANCE \
--sql='SELECT * FROM regional_sales LIMIT 5'

Os dados importados da tabela do Databricks vão aparecer.

6. Limpeza

Limpar o Spanner

Excluir o banco de dados e a instância do Spanner

gcloud spanner instances delete $SPANNER_INSTANCE

Limpar o GCS

Excluir o bucket do GCS criado para hospedar os dados

gcloud storage rm --recursive gs://$GCS_BUCKET_NAME

Limpar o Databricks

Excluir catálogo/esquema/tabela

Fazer login na instância do Databricks
Clique em no menu lateral esquerdo.
Selecione o retl_tpch_project criado anteriormente na lista do catálogo.

Na lista "Esquema", selecione tpch_data que foi criado.
Selecione o regional_sales_csv criado anteriormente na lista de tabelas.
Clique em para expandir as opções da tabela e selecione Excluir.
Clique em Excluir na caixa de diálogo de confirmação para excluir a tabela.
Depois que a tabela for excluída, você vai voltar para a página de esquema.
Clique em para expandir as opções de esquema e selecione Excluir.
Clique em Excluir na caixa de diálogo de confirmação para excluir o esquema.
Depois que o esquema for excluído, você vai voltar para a página do catálogo.
Siga as etapas de 4 a 11 novamente para excluir o esquema default, se houver um.
Na página do catálogo, clique em para expandir as opções e selecione Excluir.
Clique em Excluir na caixa de diálogo de confirmação para excluir o catálogo.

Excluir local / credenciais de dados externos

Na tela "Catálogo", clique em .
Se você não encontrar a opção External Data, talvez encontre External Location em um menu suspenso Connect.
Clique no local de dados externos retl-gcs-location criado anteriormente.
Na página do local externo, clique em para expandir as opções de local e selecione Delete.
Clique em Excluir na caixa de diálogo de confirmação para excluir o local externo.
Clique em .
Clique no retl-gcs-credential criado anteriormente.
Na página de credenciais, clique em para expandir as opções e selecione Delete.
Clique em Excluir na caixa de diálogo de confirmação para excluir as credenciais.

7. Parabéns

Parabéns por concluir o codelab.

O que aprendemos

Como carregar dados no Databricks
Como criar um bucket do GCS
Como exportar uma tabela do Databricks para o GCS no formato CSV
Como configurar uma instância do Spanner
Como carregar tabelas CSV no Spanner com o Dataflow