Carregar dados do Cloud Storage no BigQuery usando o Workflows

Prepare o ambiente

Para preparar seu ambiente, crie um banco de dados do Firestore, clone exemplos de código do repositório do GitHub, criar recursos usando Terraform editar o arquivo YAML do Workflows e instalar os requisitos do gerador de arquivos.

1. Para criar um banco de dados do Firestore, faça o seguinte:

   1. No Console do Google Cloud, acesse a página Firestore.

      Vá para o Firestore

   2. Clique em Selecionar modo nativo.

   3. No menu Selecionar um local, selecione a região em que você quer hospedar o banco de dados do Firestore. Recomendamos escolher uma região próxima à sua localização física.

   4. Clique em Create database.

2. No Cloud Shell, clone o repositório de origem:

   cd $HOME && git clone https://github.com/GoogleCloudPlatform/workflows-demos
   cd workflows-demos/workflows-bigquery-load
   

3. No Cloud Shell, use o Terraform para criar os recursos a seguir:

   terraform init
   terraform apply \
       -var project_id=PROJECT_ID \
       -var region=REGION \
       -var zone=ZONE \
       --auto-approve
   

   Substitua:

   • PROJECT_ID: é o ID do projeto do Google Cloud.
   • REGION: um local geográfico específico do Google Cloud para hospedar seus recursos, por exemplo, us-central1
   • ZONE: um local dentro de uma região para hospedar seus recursos, por exemplo, us-central1-b

   Você verá uma mensagem semelhante a esta: Apply complete! Resources: 7 added, 0 changed, 1 destroyed.

   O Terraform pode ajudar você a criar, alterar e fazer upgrade da infraestrutura em escala com segurança e previsibilidade. Os seguintes recursos são criados no seu projeto:

   • Contas de serviço com os privilégios necessários para garantir o acesso seguro aos recursos.
   • Um conjunto de dados do BigQuery chamado serverless_elt_dataset e uma tabela chamada word_count para carregar os arquivos de entrada.
   • Um bucket do Cloud Storage chamado ${project_id}-ordersbucket para preparar arquivos de entrada.
   • As cinco funções do Cloud Run a seguir:
     • file_add_handler adiciona o nome dos arquivos adicionados ao bucket do Cloud Storage à coleção do Firestore.
     • create_job cria um novo job de carregamento do BigQuery e associa arquivos na coleção do Firebase ao job.
     • create_query cria um novo job de consulta do BigQuery.
     • poll_bigquery_job recebe o status de um job do BigQuery.
     • run_bigquery_job inicia um job do BigQuery

4. Encontre os URLs das funções do Cloud Run create_job, create_query, poll_job e run_bigquery_job que você implantou na etapa anterior.

   gcloud functions describe create_job | grep url
   gcloud functions describe poll_bigquery_job | grep url
   gcloud functions describe run_bigquery_job | grep url
   gcloud functions describe create_query | grep url
   

   O resultado será assim:

   url: https://REGION-PROJECT_ID.cloudfunctions.net/create_job
   url: https://REGION-PROJECT_ID.cloudfunctions.net/poll_bigquery_job
   url: https://REGION-PROJECT_ID.cloudfunctions.net/run_bigquery_job
   url: https://REGION-PROJECT_ID.cloudfunctions.net/create_query
   

   Anote esses URLs, porque eles serão necessários quando você implantar seu fluxo de trabalho.

Criar e implantar um fluxo de trabalho

1. No Cloud Shell, abra o arquivo de origem do fluxo de trabalho workflow.yaml:

   main:
     steps:
       - constants:
           assign:
             - create_job_url: CREATE_JOB_URL
             - poll_job_url: POLL_BIGQUERY_JOB_URL
             - run_job_url: RUN_BIGQUERY_JOB_URL
             - create_query_url: CREATE_QUERY_URL
             - region: BQ_REGION
             - table_name: BQ_DATASET_TABLE_NAME
           next: createJob
   
       - createJob:
           call: http.get
           args:
             url: ${create_job_url}
             auth:
                 type: OIDC
             query:
                 region: ${region}
                 table_name: ${table_name}
           result: job
           next: setJobId
   
       - setJobId:
           assign:
             - job_id: ${job.body.job_id}
           next: jobCreateCheck
   
       - jobCreateCheck:
           switch:
             - condition: ${job_id == Null}
               next: noOpJob
           next: runLoadJob
   
       - runLoadJob:
           call: runBigQueryJob
           args:
               job_id: ${job_id}
               run_job_url: ${run_job_url}
               poll_job_url: ${poll_job_url}
           result: jobStatus
           next: loadRunCheck
   
       - loadRunCheck:
           switch:
             - condition: ${jobStatus == 2}
               next: createQueryJob
           next: failedLoadJob
   
       - createQueryJob:
           call: http.get
           args:
             url: ${create_query_url}
             query:
                 qs: "select count(*) from serverless_elt_dataset.word_count"
                 region: "US"
             auth:
                 type: OIDC
           result: queryjob
           next: setQueryJobId
   
       - setQueryJobId:
           assign:
             - qid: ${queryjob.body.job_id}
           next: queryCreateCheck
   
       - queryCreateCheck:
           switch:
             - condition: ${qid == Null}
               next: failedQueryJob
           next: runQueryJob
   
       - runQueryJob:
           call: runBigQueryJob
           args:
             job_id: ${qid}
             run_job_url: ${run_job_url}
             poll_job_url: ${poll_job_url}
           result: queryJobState
           next: runQueryCheck
   
       - runQueryCheck:
           switch:
             - condition: ${queryJobState == 2}
               next: allDone
           next: failedQueryJob
   
       - noOpJob:
           return: "No files to import"
           next: end
   
       - allDone:
           return: "All done!"
           next: end
   
       - failedQueryJob:
           return: "Query job failed"
           next: end
   
       - failedLoadJob:
           return: "Load job failed"
           next: end
   
   
   runBigQueryJob:
     params: [job_id, run_job_url, poll_job_url]
     steps:
       - startBigQueryJob:
           try:
             call: http.get
             args:
                 url: ${run_job_url}
                 query:
                   job_id: ${job_id}
                 auth:
                   type: OIDC
                 timeout: 600
             result: submitJobState
           retry: ${http.default_retry}
           next: validateSubmit
   
       - validateSubmit:
           switch:
             - condition: ${submitJobState.body.status == 1}
               next: sleepAndPollLoad
           next: returnState
   
       - returnState:
           return: ${submitJobState.body.status}
   
       - sleepAndPollLoad:
           call: sys.sleep
           args:
             seconds: 5
           next: pollJob
   
       - pollJob:
           try:
             call: http.get
             args:
               url: ${poll_job_url}
               query:
                 job_id: ${job_id}
               auth:
                 type: OIDC
               timeout: 600
             result: pollJobState
           retry:
             predicate: ${http.default_retry_predicate}
             max_retries: 10
             backoff:
               initial_delay: 1
               max_delay: 60
               multiplier: 2
           next: stateCheck
   
       - stateCheck:
           switch:
             - condition: ${pollJobState.body.status == 2}
               return: ${pollJobState.body.status}
             - condition: ${pollJobState.body.status == 3}
               return: ${pollJobState.body.status}
           next: sleepAndPollLoad

   Substitua:

   • CREATE_JOB_URL: o URL da função para criar um novo job
   • POLL_BIGQUERY_JOB_URL: o URL da função para pesquisar o status de um job em execução
   • RUN_BIGQUERY_JOB_URL: o URL da função para iniciar um job de carga do BigQuery.
   • CREATE_QUERY_URL: o URL da função para iniciar um job de consulta do BigQuery.
   • BQ_REGION: o Região do BigQuery onde os dados são armazenados, por exemplo, US
   • BQ_DATASET_TABLE_NAME: o BigQuery nome da tabela do conjunto de dados no formato PROJECT_ID.serverless_elt_dataset.word_count

2. Implante o arquivo workflow:

   gcloud workflows deploy WORKFLOW_NAME \
       --location=WORKFLOW_REGION \
       --description='WORKFLOW_DESCRIPTION' \
       --service-account=workflow-runner@PROJECT_ID.iam.gserviceaccount.com \
       --source=workflow.yaml
   

   Substitua:

   • WORKFLOW_NAME: o nome exclusivo do fluxo de trabalho.
   • WORKFLOW_REGION: o região em que o fluxo de trabalho é implantado, por exemplo, us-central1
   • WORKFLOW_DESCRIPTION: a descrição do fluxo de trabalho

3. Crie um ambiente virtual em Python 3 e requisitos de instalação para o gerador de arquivos:

   sudo apt-get install -y python3-venv
   python3 -m venv env
   . env/bin/activate
   cd generator
   pip install -r requirements.txt
   

Gerar arquivos para importação

O script Python gen.py gera conteúdo aleatório em Formato Avro. O esquema é o mesmo que na tabela word_count do BigQuery. Esses arquivos Avro são copiados no bucket especificado do Cloud Storage.

No Cloud Shell, gere os arquivos:

python gen.py -p PROJECT_ID \
    -o PROJECT_ID-ordersbucket \
    -n RECORDS_PER_FILE \
    -f NUM_FILES \
    -x FILE_PREFIX

Substitua:

• RECORDS_PER_FILE: o número de registros em um único arquivo
• NUM_FILES: o número total de arquivos a serem enviados
• FILE_PREFIX: o prefixo dos nomes dos arquivos gerados.

Ver entradas de arquivo no Firestore

Quando os arquivos são copiados para o Cloud Storage, A função handle_new_file do Cloud Run é acionada. Essa função adiciona a lista de arquivos à matriz da lista de arquivos no documento new na coleção jobs do Firestore.

Para conferir a lista de arquivos, acesse o console do Google Cloud Página Dados do Firestore.

Ir para Dados

Acione o fluxo de trabalho

Os Workflows vinculam uma série de tarefas sem servidor a partir dos serviços do Google Cloud e da API. Etapas individuais desse fluxo de trabalho são executadas como as funções do Cloud Run e o estado é armazenado no Firestore. Tudo as chamadas para as funções do Cloud Run são autenticadas usando a conta de serviço do fluxo de trabalho.

No Cloud Shell, execute o fluxo de trabalho:

gcloud workflows execute WORKFLOW_NAME

No diagrama a seguir, mostramos as etapas usadas no fluxo de trabalho:

O fluxo de trabalho é dividido em duas partes: o principal e o sub-fluxo. O fluxo de trabalho principal processa a criação de jobs e a execução condicional enquanto o subfluxo de trabalho executa um job do BigQuery. O fluxo de trabalho executa as seguintes operações:

• A função create_job do Cloud Run cria um novo objeto de job. recebe a lista de arquivos adicionados ao Cloud Storage pelo documento do Firestore e associa os arquivos à carga trabalho. Se não houver arquivos para carregar, a função não criará um novo job.
• A função create_query do Cloud Run usa a consulta que precisa seja executado com a região do BigQuery em que a consulta deve ser executado. A função cria o job no Firestore e retorna o ID dele.
• A função run_bigquery_job do Cloud Run recebe o ID do job que precisa ser executado e, em seguida, chama a API BigQuery para enviar o job.
• Em vez de esperar que o trabalho seja concluído função do Cloud Run, é possível pesquisar periodicamente o status do job.
  • A função poll_bigquery_job do Cloud Run fornece a status do trabalho. É chamado repetidamente até que o job seja concluído.
  • Para adicionar um atraso entre as chamadas para o poll_bigquery_job função do Cloud Run, Rotina sleep é chamado no Workflows.

Ver o status do job

É possível ver a lista de arquivos e o status do job.

1. Na Console do Google Cloud, acesse a página Dados do Firestore.

   Ir para Dados

2. Um identificador exclusivo (UUID) é gerado para cada job. Para visualizar job_type e status, clique no ID do job. Cada job pode ter um dos seguintes tipos e status:

   • job_type: o tipo de job que está sendo executado pelo fluxo de trabalho com um dos seguintes valores:

     • 0: carregar dados no BigQuery.
     • 1: execute uma consulta no BigQuery.

   • status: o estado atual do job com um dos seguintes valores:

     • 0: o job foi criado, mas não foi iniciado.
     • 1: o job está em execução.
     • 2: o job concluiu a execução com êxito.
     • 3: ocorreu um erro, e o job não foi concluído.

   O objeto do trabalho também contém atributos de metadados, como a região do conjunto de dados do BigQuery, e, se for um job de consulta, a string de consulta em execução.

Ver dados no BigQuery

Para confirmar se o job de ELT foi bem-sucedido, verifique se os dados aparecem no tabela.

1. No console do Google Cloud, acesse a página Editor do BigQuery.

   Acessar o Editor

2. Clique na tabela serverless_elt_dataset.word_count.

3. Clique na guia Visualizar.

   

Programe o fluxo de trabalho

Para executar periodicamente o fluxo de trabalho em uma programação, use o Cloud Scheduler.