Caricare i dati da Cloud Storage in BigQuery utilizzando Workflows

Last reviewed 2021-05-12 UTC
Questo tutorial mostra come eseguire in modo affidabile workflow serverless utilizzando Workflows, Cloud Run Functions, e Firestore per caricare dati non elaborati, come i log degli eventi, da Cloud Storage a BigQuery. Le piattaforme di analisi in genere dispongono di uno strumento di orchestrazione per caricare periodicamente i dati in BigQuery utilizzando i job BigQuery, e poi trasformare i dati per fornire metriche aziendali utilizzando istruzioni SQL, incluse le istruzioni del linguaggio procedurale BigQuery. Questo tutorial è rivolto a sviluppatori e architetti che vogliono creare pipeline di elaborazione dei dati serverless basate sugli eventi. Il tutorial presuppone che tu abbia familiarità con YAML, SQL e Python.

Architettura

Il seguente diagramma mostra l'architettura di alto livello di una pipeline di estrazione, caricamento e trasformazione (ELT) serverless che utilizza Workflows.

Pipeline di estrazione, caricamento e trasformazione.

Nel diagramma precedente, considera una piattaforma di vendita al dettaglio che raccoglie periodicamente gli eventi di vendita come file da vari negozi e poi scrive i file in un bucket Cloud Storage. Gli eventi vengono utilizzati per fornire metriche aziendali mediante l'importazione e l'elaborazione in BigQuery. Questa architettura fornisce un sistema di orchestrazione serverless e affidabile per importare i file in BigQuery ed è suddivisa nei due moduli seguenti:

  • Elenco file: gestisce l'elenco dei file non elaborati aggiunti a un bucket Cloud Storage in una raccolta Firestore. Questo modulo funziona tramite una funzione Cloud Run attivata da un evento di archiviazione Object Finalize, generato quando viene aggiunto un nuovo file al bucket Cloud Storage. Il nome file viene aggiunto all'array files della raccolta denominata new in Firestore.

  • Workflow: esegue i workflow pianificati. Cloud Scheduler attiva un workflow che esegue una serie di passaggi in base a una sintassi basata su YAML per orchestrare il caricamento e poi la trasformazione dei dati in BigQuery chiamando Cloud Run Functions. I passaggi del workflow chiamano Cloud Run Functions per eseguire le seguenti attività:

    • Crea e avvia un job di caricamento BigQuery.
    • Esegue il polling dello stato del job di caricamento.
    • Crea e avvia il job di query di trasformazione.
    • Esegue il polling dello stato del job di trasformazione.

L'utilizzo delle transazioni per gestire l'elenco dei nuovi file in Firestore contribuisce a garantire che nessun file venga perso quando un workflow li importa in BigQuery. Le esecuzioni separate del workflow vengono rese idempotenti memorizzando i metadati e lo stato del job in Firestore.

Obiettivi

  • Crea un database Firestore.
  • Configura un trigger di Cloud Run Functions per monitorare i file aggiunti al bucket Cloud Storage in Firestore.
  • Esegui il deployment di Cloud Run Functions per eseguire e monitorare i job BigQuery.
  • Esegui il deployment e l'esecuzione di un workflow per automatizzare il processo.

Costi

In questo documento vengono utilizzati i seguenti componenti fatturabili di Google Cloud:

Per generare una stima dei costi in base all'utilizzo previsto, utilizza il calcolatore prezzi.

I nuovi Google Cloud utenti potrebbero avere diritto a una prova senza costi.

Al termine delle attività descritte in questo documento, puoi evitare l'addebito di ulteriori costi eliminando le risorse che hai creato. Per saperne di più, consulta Esegui la pulizia.

Prima di iniziare

  1. Nella Google Cloud console, nella pagina di selezione del progetto, seleziona o crea un Google Cloud progetto.

    Ruoli richiesti per selezionare o creare un progetto

    • Seleziona un progetto: la selezione di un progetto non richiede un ruolo IAM specifico. Puoi selezionare qualsiasi progetto su cui ti è stato concesso un ruolo.
    • Crea un progetto: per creare un progetto, devi disporre del ruolo Autore progetto (roles/resourcemanager.projectCreator), che contiene l' resourcemanager.projects.create autorizzazione. Scopri come concedere i ruoli.

    Vai al selettore di progetti

  2. Verifica che la fatturazione sia abilitata per il tuo Google Cloud progetto.

  3. Abilita le API Cloud Build, Cloud Run Functions, Identity and Access Management, Resource Manager e Workflows.

    Ruoli richiesti per abilitare le API

    Per abilitare le API, devi disporre del ruolo IAM Service Usage Admin (roles/serviceusage.serviceUsageAdmin), che contiene l'autorizzazione serviceusage.services.enable. Scopri come concedere i ruoli.

    Abilita le API

  4. Vai alla pagina di benvenuto e prendi nota dell'ID progetto da utilizzare in un passaggio successivo.

    Vai alla pagina di benvenuto

  5. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

Prepara l'ambiente

Per preparare l'ambiente, crea un database Firestore, clona gli esempi di codice dal repository GitHub, crea le 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 Google Cloud console, vai alla pagina Firestore.

      Vai a Firestore

    2. Fai clic su Seleziona modalità nativa.

    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: l'ID Google Cloud progetto
    • REGION: una località geografica specifica per ospitare le risorse, ad esempio us-central1 Google Cloud
    • ZONE: una località all'interno di una regione per ospitare le 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 progetto vengono create le seguenti risorse:

    • Service account con i privilegi richiesti per garantire l'accesso sicuro alle risorse.
    • Un set di dati BigQuery denominato serverless_elt_dataset e una tabella denominata word_count per caricare i file in entrata.
    • Un bucket Cloud Storage denominato ${project_id}-ordersbucket per l'organizzazione dei file di input.
    • Le seguenti cinque Cloud Run Functions:
      • 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 di Cloud Run Functions 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é sono necessari quando esegui il deployment del workflow.

Crea ed esegui il deployment di un flusso di lavoro

  1. In Cloud Shell, apri il file di origine del workflow, 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 sono 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 workflow
    • WORKFLOW_REGION: la regione in cui viene eseguito il deployment del workflow, 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

Genera i file da importare

Lo script Python gen.py genera contenuti casuali in formato Avro. Lo schema è lo stesso della tabella word_count di 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

Visualizza 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 jobs di Firestore.

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

Vai a Dati

Elenco dei file aggiunti alla raccolta.

Attiva il workflow

Workflows collega una serie di attività serverless da Google Cloud e servizi basati sulle API. I singoli passaggi di questo workflow vengono eseguiti come Cloud Run Functions e lo stato viene archiviato in Firestore. Tutte le chiamate a Cloud Run Functions vengono autenticate utilizzando il account di servizio del workflow.

In Cloud Shell, esegui il workflow:

gcloud workflows execute WORKFLOW_NAME

Il seguente diagramma mostra i passaggi utilizzati nel workflow:

Passaggi utilizzati nel flusso di lavoro principale e secondario.

Il workflow è suddiviso in due parti: il workflow principale e il subworkflow. Il workflow principale gestisce la creazione dei job e l'esecuzione condizionale, mentre il sub workflow esegue un job BigQuery. Il workflow 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 query. 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 in Cloud Run Functions, puoi eseguire periodicamente il polling dello stato del job.
    • La funzione Cloud Run poll_bigquery_job fornisce lo stato del job. Viene chiamata ripetutamente fino al completamento del job.
    • Per aggiungere un ritardo tra le chiamate alla poll_bigquery_job funzione Cloud Run, una sleep routine viene chiamata da Workflows.

Visualizza lo stato del job

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

  1. Nella Google Cloud console, 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 workflow con uno dei seguenti valori:

      • 0: carica i dati in BigQuery.
      • 1: esegui una query in BigQuery.

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

      • 0: il job è stato creato, ma non è stato avviato.
      • 1: il job è in esecuzione.
      • 2: il job ha completato l'esecuzione 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.

Elenco dei file con lo stato del job evidenziato.

Visualizza dati in BigQuery

Per verificare che il job ELT sia stato completato correttamente, controlla che i dati vengano visualizzati nella tabella.

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

    Vai all'editor

  2. Fai clic sulla tabella serverless_elt_dataset.word_count.

  3. Fai clic sulla scheda Anteprima.

    Scheda Anteprima che mostra i dati nella tabella.

Pianifica il workflow

Per eseguire periodicamente il workflow in base a una pianificazione, puoi utilizzare Cloud Scheduler.

Libera spazio

Il modo più semplice per eliminare la fatturazione è quello di eliminare il Google Cloud progetto che hai creato per il tutorial. In alternativa, puoi eliminare le singole risorse.

Elimina le singole risorse

  1. In Cloud Shell, rimuovi tutte le risorse create utilizzando Terraform:

    cd $HOME/bigquery-workflows-load
terraform destroy \
-var project_id=PROJECT_ID \
-var region=REGION \
-var zone=ZONE \
--auto-approve

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

    Vai a Dati

  3. Accanto a Job, fai clic su Menu e seleziona Elimina.

    Percorso del menu per eliminare una raccolta.

Elimina il progetto

  1. Nella Google Cloud console, vai alla pagina Gestisci risorse.

    Vai a Gestisci risorse

  2. Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare, quindi fai clic su Elimina.
  3. Nella finestra di dialogo, digita l'ID progetto e fai clic su Chiudi per eliminare il progetto.

Passaggi successivi