Caricare i dati da Cloud Storage in BigQuery utilizzando Workflows

prepara l'ambiente

Per preparare l'ambiente, crea un database Firestore, clona gli esempi di codice dal repository GitHub, crea risorse utilizzando Terraform, modifica il file YAML di Workflows e installa i requisiti per il generatore di file.

1. Per creare un database Firestore:

   1. Nella console Google Cloud , vai alla pagina Firestore.

      Vai a Firestore

   2. Fai clic su Seleziona modalità Native.

   3. Nel menu Seleziona una località, seleziona la regione in cui vuoi ospitare il database Firestore. Ti consigliamo di scegliere una regione vicina alla tua posizione fisica.

   4. Fai clic su Crea database.

2. In Cloud Shell, clona il repository di origine:

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

3. In Cloud Shell, crea le seguenti risorse utilizzando Terraform:

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

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo ID progetto Google Cloud
   • REGION: una località geografica specifica Google Cloud in cui ospitare le tue risorse, ad esempio us-central1
   • ZONE: una località all'interno di una regione in cui ospitare le tue risorse, ad esempio us-central1-b

   Dovresti visualizzare un messaggio simile al seguente: Apply complete! Resources: 7 added, 0 changed, 1 destroyed.

   Terraform può aiutarti a creare, modificare e aggiornare l'infrastruttura su larga scala in modo sicuro e prevedibile. Nel tuo progetto vengono create le seguenti risorse:

   • Service account con i privilegi necessari per garantire l'accesso sicuro alle tue risorse.
   • Un set di dati BigQuery denominato serverless_elt_dataset e una tabella denominata word_count per caricare i file in arrivo.
   • Un bucket Cloud Storage denominato ${project_id}-ordersbucket per la gestione temporanea dei file di input.
   • Le seguenti cinque funzioni Cloud Run:
     • file_add_handler aggiunge il nome dei file aggiunti al bucket Cloud Storage alla raccolta Firestore.
     • create_job crea un nuovo job di caricamento BigQuery e associa i file nella raccolta Firebase al job.
     • create_query crea un nuovo job di query BigQuery.
     • poll_bigquery_job recupera lo stato di un job BigQuery.
     • run_bigquery_job avvia un job BigQuery.

4. Recupera gli URL delle funzioni Cloud Run create_job, create_query, poll_job e run_bigquery_job di cui hai eseguito il deployment nel passaggio precedente.

   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
   

   L'output è simile al seguente:

   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
   

   Prendi nota di questi URL perché ti serviranno quando implementerai il workflow.

Crea ed esegui il deployment di un flusso di lavoro

1. In Cloud Shell, apri il file di origine del flusso di lavoro, 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

   Sostituisci quanto segue:

   • CREATE_JOB_URL: l'URL della funzione per creare un nuovo job
   • POLL_BIGQUERY_JOB_URL: l'URL della funzione per eseguire il polling dello stato di un job in esecuzione
   • RUN_BIGQUERY_JOB_URL: l'URL della funzione per avviare un job di caricamento BigQuery
   • CREATE_QUERY_URL: l'URL della funzione per avviare un job di query BigQuery
   • BQ_REGION: la regione BigQuery in cui vengono archiviati i dati, ad esempio US
   • BQ_DATASET_TABLE_NAME: il nome della tabella del set di dati BigQuery nel formato PROJECT_ID.serverless_elt_dataset.word_count

2. Esegui il deployment del file workflow:

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

   Sostituisci quanto segue:

   • WORKFLOW_NAME: il nome univoco del flusso di lavoro
   • WORKFLOW_REGION: la regione in cui viene eseguito il deployment del flusso di lavoro, ad esempio us-central1
   • WORKFLOW_DESCRIPTION: la descrizione del workflow

3. Crea un ambiente virtuale Python 3 e installa i requisiti per il generatore di file:

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

Generare file da importare

Lo script Python gen.py genera contenuti casuali in formato Avro. Lo schema è lo stesso della tabella word_count BigQuery. Questi file Avro vengono copiati nel bucket Cloud Storage specificato.

In Cloud Shell, genera i file:

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

Sostituisci quanto segue:

• RECORDS_PER_FILE: il numero di record in un singolo file
• NUM_FILES: il numero totale di file da caricare
• FILE_PREFIX: il prefisso per i nomi dei file generati

Visualizzare le voci dei file in Firestore

Quando i file vengono copiati in Cloud Storage, viene attivata la funzione Cloud Run handle_new_file. Questa funzione aggiunge l'elenco dei file all'array dell'elenco dei file nel documento new nella raccolta Firestore jobs.

Per visualizzare l'elenco dei file, nella console Google Cloud , vai alla pagina Dati di Firestore.

Vai a Dati

Attivare il workflow

Workflows collega una serie di attività serverless daGoogle Cloud e servizi API. I singoli passaggi di questo flusso di lavoro vengono eseguiti come funzioni Cloud Run e lo stato viene archiviato in Firestore. Tutte le chiamate alle funzioni Cloud Run vengono autenticate utilizzando l'account di servizio del flusso di lavoro.

In Cloud Shell, esegui il flusso di lavoro:

gcloud workflows execute WORKFLOW_NAME

Il seguente diagramma mostra i passaggi utilizzati nel flusso di lavoro:

Il flusso di lavoro è suddiviso in due parti: il flusso di lavoro principale e il flusso di lavoro secondario. Il workflow principale gestisce la creazione dei job e l'esecuzione condizionale, mentre il workflow secondario esegue un job BigQuery. Il flusso di lavoro esegue le seguenti operazioni:

• La funzione Cloud Run create_job crea un nuovo oggetto job, recupera l'elenco dei file aggiunti a Cloud Storage dal documento Firestore e associa i file al job di caricamento. Se non ci sono file da caricare, la funzione non crea un nuovo job.
• La funzione Cloud Run create_query accetta la query da eseguire insieme alla regione BigQuery in cui deve essere eseguita. La funzione crea il job in Firestore e restituisce l'ID job.
• La funzione Cloud Run run_bigquery_job recupera l'ID del job da eseguire e poi chiama l'API BigQuery per inviare il job.
• Anziché attendere il completamento del job nella funzione Cloud Run, puoi eseguire periodicamente il polling dello stato del job.
  • La funzione Cloud Run poll_bigquery_job fornisce lo stato del job. Viene chiamato ripetutamente fino al completamento del job.
  • Per aggiungere un ritardo tra le chiamate alla funzione Cloud Run, viene chiamata una routine sleep da Workflows.poll_bigquery_job

Visualizzare lo stato del job

Puoi visualizzare l'elenco dei file e lo stato del job.

1. Nella console Google Cloud , vai alla pagina Dati di Firestore.

   Vai a Dati

2. Per ogni job viene generato un identificatore univoco (UUID). Per visualizzare job_type e status, fai clic sull'ID job. Ogni job può avere uno dei seguenti tipi e stati:

   • job_type: Il tipo di job eseguito dal flusso di lavoro con uno dei seguenti valori:

     • 0: Carica i dati in BigQuery.
     • 1. Esegui una query in BigQuery.

   • status: lo stato attuale del lavoro con uno dei seguenti valori:

     • 0: Il job è stato creato, ma non avviato.
     • 1: Il job è in esecuzione.
     • 2: L'esecuzione del job è stata completata correttamente.
     • 3: Si è verificato un errore e il job non è stato completato correttamente.

   L'oggetto job contiene anche attributi di metadati come la regione del set di dati BigQuery, il nome della tabella BigQuery e, se si tratta di un job di query, la stringa di query in esecuzione.

Visualizzare i dati in BigQuery

Per verificare che il job ELT sia andato a buon fine, controlla che i dati vengano visualizzati nella tabella.

1. Nella console Google Cloud , vai alla pagina BigQuery Editor.

   Vai all'editor

2. Fai clic sulla tabella serverless_elt_dataset.word_count.

3. Fai clic sulla scheda Anteprima.

   

Pianifica il workflow

Per eseguire periodicamente il flusso di lavoro in base a una pianificazione, puoi utilizzare Cloud Scheduler.