Lavorare con le pipeline di dati Dataflow

Nota: puoi segnalare i problemi relativi alle pipeline di dati di Dataflow e richiedere nuove funzionalità all'indirizzo google-data-pipelines-feedback."

Panoramica

Puoi utilizzare le pipeline di dati Dataflow per le attività seguenti:

• Crea pianificazioni dei job ricorrenti.
• Scopri dove vengono utilizzate le risorse per più esecuzioni di job.
• Definire e gestire gli obiettivi di aggiornamento dei dati.
• Visualizza in dettaglio le singole fasi della pipeline per correggere e ottimizzare le pipeline.

Per la documentazione relativa all'API, consulta la documentazione di riferimento sulle pipeline di dati.

Funzionalità

• Crea una pipeline batch ricorrente per eseguire un job batch in base a una pianificazione.
• Crea una pipeline batch incrementale ricorrente per eseguire un job batch sulla versione più recente dei dati di input.
• Usa il prospetto di riepilogo della pipeline per visualizzare l'utilizzo aggregato di capacità e risorse di una pipeline.
• Visualizza l'aggiornamento dei dati di una pipeline in modalità flusso. Questa metrica, che si evolve nel tempo, può essere associata a un avviso che segnala quando l'aggiornamento scende al di sotto di un obiettivo specificato.
• Utilizza i grafici delle metriche della pipeline per confrontare i job di pipeline in modalità batch e trovare anomalie.

Limitazioni

• Disponibilità regionale: puoi creare pipeline di dati nelle regioni disponibili di Cloud Scheduler.

• Quota:

  • Numero predefinito di pipeline per progetto: 500

  • Numero predefinito di pipeline per organizzazione: 2500

    La quota a livello di organizzazione è disabilitata per impostazione predefinita. Puoi attivare le quote a livello di organizzazione e, in questo caso, ogni organizzazione può avere al massimo 2500 pipeline per impostazione predefinita.

  Nota: se vuoi aumentare la quota, contatta l'assistenza Google Cloud. Tuttavia, il valore limite è soggetto alle quote e ai limiti di Cloud Scheduler e di Dataflow.

• Etichette: non puoi utilizzare le etichette definite dall'utente per etichettare le pipeline di dati Dataflow. Tuttavia, quando utilizzi il campo additionalUserLabels, questi valori vengono trasmessi al job Dataflow. Per ulteriori informazioni su come le etichette vengono applicate a singoli job Dataflow, consulta Opzioni della pipeline.

Tipi di pipeline di dati

Dataflow ha due tipi di pipeline di dati: in modalità flusso e batch. Entrambi i tipi di pipeline eseguono job definiti nei modelli di Dataflow.

Pipeline di dati in modalità flusso

Una pipeline di dati in modalità flusso esegue un job di inserimento di flussi di Dataflow immediatamente dopo la creazione.

Pipeline di dati in batch

Una pipeline di dati batch esegue un job batch di Dataflow in base a una pianificazione definita dall'utente. Il nome file di input della pipeline batch può essere parametrizzato per consentire l'elaborazione incrementale della pipeline batch.

Nota: ogni nome di job batch Dataflow creato da una pipeline di dati batch utilizza il seguente pattern di denominazione: <pipeline_id>-MP--<timestamp>. Il valore di timestamp ha una granularità in secondi. La stringa -MP– è riservata per indicare che questo job batch Dataflow è stato creato da una pipeline di dati batch. Poiché i nomi dei job hanno un limite di 40 caratteri, pipeline_id viene troncato dopo 25 caratteri.

Pipeline batch incrementali

Puoi utilizzare i segnaposto di data/ora per specificare un formato di file di input incrementale per una pipeline batch.

• È possibile utilizzare i segnaposto per anno, mese, data, ora, minuto e secondo e devono essere nel formato strftime(). I segnaposto sono preceduti dal simbolo di percentuale (%).
• La formattazione dei parametri non viene verificata durante la creazione della pipeline.
  • Esempio: se specifichi "gs://bucket/Y" come percorso di input con parametri, questo viene valutato come "gs://bucket/Y", perché "Y" senza il carattere "%" precedente non viene mappato al formato strftime().

In corrispondenza di ogni data di esecuzione pianificata della pipeline batch, la porzione segnaposto del percorso di input viene valutata in base alla data/ora attuale (o in timeshift). I valori delle date vengono valutati utilizzando la data corrente nel fuso orario del job pianificato. Se il percorso valutato corrisponde al percorso di un file di input, il file viene selezionato per l'elaborazione dalla pipeline batch al momento pianificato.

• Esempio: una pipeline batch è pianificata per ripetersi all'inizio di ogni ora PST. Se parametrizzi il percorso di input come gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv, il 15 aprile 2021 alle 18:00 PST, il percorso di input viene valutato in gs://bucket-name/2021-04-15/prefix-18_00.csv.

Utilizzare i parametri di timeshift

Puoi utilizzare i parametri di spostamento temporale + o - minuti o ora. Per supportare la corrispondenza di un percorso di input con una data/ora valutata che viene spostata prima o dopo la data/ora attuale della pianificazione della pipeline, racchiudi questi parametri tra parentesi graffe. Utilizza il formato {[+|-][0-9]+[m|h]}. La pipeline batch continua a ripetersi all'orario pianificato, ma il percorso di input viene valutato con lo offset temporale specificato.

• Esempio: una pipeline batch è pianificata per ripetersi all'inizio di ogni ora PST. Se parametrizzi il percorso di input come gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv{-2h}, il 15 aprile 2021 alle 18:00 PST, il percorso di input viene valutato in gs://bucket-name/2021-04-15/prefix-16_00.csv.

Ruoli della pipeline di dati

Affinché le operazioni della pipeline di dati di Dataflow abbiano esito positivo, devi avere i ruoli IAM necessari, come segue:

1. Per eseguire le operazioni devi avere il ruolo appropriato:

   • Datapipelines.admin: può eseguire tutte le operazioni della pipeline di dati
   • Datapipelines.viewer: può visualizzare pipeline di dati e job
   • Datapipelines.invoker: può richiamare l'esecuzione di un job di pipeline di dati (questo ruolo può essere abilitato utilizzando l'API)

2. L'account di servizio utilizzato da Cloud Scheduler deve avere il ruolo roles/iam.serviceAccountUser, indipendentemente dal fatto che sia specificato dall'utente o come account di servizio Compute Engine predefinito. Per saperne di più, consulta Ruoli della pipeline di dati.

3. Per agire come account di servizio utilizzato da Cloud Scheduler e da Dataflow, devi disporre del ruolo roles/iam.serviceAccountUser per quell'account. Se non selezioni un account di servizio per Cloud Scheduler e Dataflow, viene utilizzato l'account di servizio predefinito di Compute Engine.

Nota: se riscontri uno stato di errore NOT_FOUND da Cloud Scheduler con la voce di log type.googleapis.com/google.cloud.scheduler.logging.AttemptFinished, spesso indica la mancanza di autorizzazioni per gli account di servizio per Cloud Scheduler o per Dataflow. Verifica e correggi queste autorizzazioni per risolvere i problemi delle pipeline di dati.

Crea una pipeline di dati

Puoi creare una pipeline di dati Dataflow in due modi:

1. Importa un job oppure
2. Crea una pipeline di dati

Pagina di configurazione delle pipeline di dati: la prima volta che accedi alla funzionalità delle pipeline di Dataflow nella console Google Cloud, viene aperta una pagina di configurazione. Abilitare le API elencate per creare pipeline di dati.

Importa un job

Puoi importare un job Dataflow in batch o in modalità flusso basato su un modello classico o flessibile e renderlo una pipeline di dati.

1. Nella console Google Cloud, vai alla pagina Job di Dataflow.

   Vai a Job

2. Seleziona un job completato, quindi nella pagina Dettagli job seleziona +Importa come pipeline.

3. Nella pagina Crea pipeline da modello, i parametri vengono completati con le opzioni del job importato.

4. Per un job batch, nella sezione Pianifica la pipeline, fornisci una pianificazione della ricorrenza. L'inserimento di un indirizzo email per Cloud Scheduler, utilizzato per pianificare le esecuzioni batch, è facoltativo. Se non è specificato, viene utilizzato l'account di servizio Compute Engine predefinito.

Crea una pipeline di dati

1. Nella console Google Cloud, vai alla pagina pipeline di dati di Dataflow.

   Vai a Pipeline di dati

2. Seleziona + Crea pipeline di dati.

3. Nella pagina Crea pipeline da modello, specifica il nome della pipeline e compila gli altri campi di selezione e parametri dei modelli.

4. Per un job batch, nella sezione Pianifica la pipeline, fornisci una pianificazione della ricorrenza. L'inserimento di un indirizzo email per Cloud Scheduler, utilizzato per pianificare le esecuzioni batch, è facoltativo. Se non viene specificato un valore, viene utilizzato l'account di servizio Compute Engine predefinito.

Crea una pipeline di dati batch

Per creare questa pipeline di dati batch di esempio, devi avere accesso alle seguenti risorse nel tuo progetto:

• Un bucket Cloud Storage per archiviare i file di input e di output
• Un set di dati BigQuery per creare una tabella.

Questa pipeline di esempio utilizza il modello di pipeline di batch Cloud Storage Text to BigQuery. Questo modello legge i file in formato CSV da Cloud Storage, esegue una trasformazione e inserisce i valori in una tabella BigQuery con tre colonne.

1. Crea i seguenti file nell'unità locale:

   1. Un file bq_three_column_table.json contenente il seguente schema della tabella BigQuery di destinazione.

      {
        "BigQuery Schema": [
          {
            "name": "col1",
            "type": "STRING"
          },
          {
            "name": "col2",
            "type": "STRING"
          },
          {
            "name": "col3",
            "type": "INT64"
          }
        ]
      }
      

   2. Un file JavaScript split_csv_3cols.js, che implementa una semplice trasformazione dei dati di input prima dell'inserimento in BigQuery.

      function transform(line) {
          var values = line.split(',');
          var obj = new Object();
          obj.col1 = values[0];
          obj.col2 = values[1];
          obj.col3 = values[2];
          var jsonString = JSON.stringify(obj);
          return jsonString;
      }
      

   3. Un file CSV file01.csv con diversi record inseriti nella tabella BigQuery.

      b8e5087a,74,27531
      7a52c051,4a,25846
      672de80f,cd,76981
      111b92bf,2e,104653
      ff658424,f0,149364
      e6c17c75,84,38840
      833f5a69,8f,76892
      d8c833ff,7d,201386
      7d3da7fb,d5,81919
      3836d29b,70,181524
      ca66e6e5,d7,172076
      c8475eb6,03,247282
      558294df,f3,155392
      737b82a8,c7,235523
      82c8f5dc,35,468039
      57ab17f9,5e,480350
      cbcdaf84,bd,354127
      52b55391,eb,423078
      825b8863,62,88160
      26f16d4f,fd,397783
      

2. Utilizza il comando gcloud storage cp per copiare i file nelle cartelle di un bucket Cloud Storage nel tuo progetto, come segue:

   1. Copia bq_three_column_table.json e split_csv_3cols.js in gs://BUCKET_ID/text_to_bigquery/

      gcloud storage cp bq_three_column_table.json gs://BUCKET_ID/text_to_bigquery/
      gcloud storage cp split_csv_3cols.js gs://BUCKET_ID/text_to_bigquery/
      

   2. Copia file01.csv in gs://BUCKET_ID/inputs/

      gcloud storage cp file01.csv gs://BUCKET_ID/inputs/
      

3. Nella console Google Cloud, vai alla pagina Bucket di Cloud Storage.

   Vai a Bucket

4. Per creare una cartella tmp nel bucket Cloud Storage, seleziona il nome della cartella per aprire la pagina Dettagli bucket, quindi fai clic su Crea cartella.

   

5. Nella console Google Cloud, vai alla pagina pipeline di dati di Dataflow.

   Vai a Pipeline di dati

6. Seleziona Crea pipeline di dati. Inserisci o seleziona i seguenti elementi nella pagina Crea pipeline da modello:

   1. In Nome pipeline, inserisci text_to_bq_batch_data_pipeline.
   2. Per Endpoint a livello di regione, seleziona una regione di Compute Engine. Le regioni di origine e di destinazione devono corrispondere. Di conseguenza, il bucket Cloud Storage e la tabella BigQuery devono trovarsi nella stessa regione.

   3. Per Modello Dataflow, in Elabora dati in blocco (batch), seleziona File di testo su Cloud Storage in BigQuery.

      Nota: non selezionare la pipeline in modalità flusso con lo stesso nome in Elabora i dati continuamente (flusso).

   4. Per Pianifica la pipeline, seleziona una pianificazione, ad esempio Ogni ora al minuto 25, nel tuo fuso orario. Puoi modificare la pianificazione dopo aver inviato la pipeline. Fornire un indirizzo email per Cloud Scheduler, utilizzato per pianificare le esecuzioni batch, è facoltativo. Se non è specificato, viene utilizzato l'account di servizio Compute Engine predefinito.

   5. In Parametri obbligatori, inserisci quanto segue:

      1. Per il percorso delle funzioni definite dall'utente JavaScript in Cloud Storage:

         gs://BUCKET_ID/text_to_bigquery/split_csv_3cols.js
         

      2. Per Percorso JSON:

         BUCKET_ID/text_to_bigquery/bq_three_column_table.json
         

      3. Per Nome funzione JavaScript JavaScript: transform
      4. Per la tabella di output BigQuery:

         PROJECT_ID:DATASET_ID.three_column_table
         

      5. Per Percorso di input di Cloud Storage:

         BUCKET_ID/inputs/file01.csv
         

      6. Per la directory BigQuery temporanea:

         BUCKET_ID/tmp
         

      7. Per la Posizione temporanea:

         BUCKET_ID/tmp
         

   6. Fai clic su Crea pipeline.

7. Conferma le informazioni della pipeline e del modello e visualizza la cronologia attuale e precedente dalla pagina Dettagli della pipeline.

   

   Puoi modificare la pianificazione della pipeline di dati dal riquadro Informazioni pipeline nella pagina Dettagli pipeline. {: width="375"}

Puoi anche eseguire una pipeline batch on demand utilizzando il pulsante Esegui nella console di Dataflow Pipelines.

Crea una pipeline di dati in modalità flusso di esempio

Puoi creare una pipeline di dati in modalità flusso di esempio seguendo le istruzioni della pipeline batch di esempio, con le seguenti differenze:

• Per Pianificazione della pipeline, non specificare una pianificazione per una pipeline di dati in modalità flusso. Il job di inserimento di flussi di Dataflow viene avviato immediatamente.
• Per Modello Dataflow, in Elabora i dati in modo continuo (flusso), seleziona File di testo su Cloud Storage in BigQuery.
• Per il Tipo di macchina worker, la pipeline elabora l'insieme iniziale di file corrispondenti al pattern gs://BUCKET_ID/inputs/file01.csv e gli eventuali file aggiuntivi corrispondenti a questo pattern che carichi nella cartella inputs/. Se le dimensioni dei file CSV superano diversi GB, per evitare possibili errori di esaurimento della memoria, seleziona un tipo di macchina con memoria più elevata rispetto al tipo di macchina n1-standard-4 predefinito, ad esempio n1-highmem-8.

Risoluzione dei problemi

Questa sezione mostra come risolvere i problemi relativi alle pipeline di dati di Dataflow.

Impossibile avviare il job della pipeline di dati

Quando utilizzi pipeline di dati per creare una pianificazione di job ricorrente, il job Dataflow potrebbe non essere avviato e potrebbe essere visualizzato un errore di stato 503 nei file di log di Cloud Scheduler.

Questo problema si verifica quando Dataflow non è temporaneamente in grado di eseguire il job.

Per risolvere questo problema, configura Cloud Scheduler in modo da riprovare il job. Poiché il problema è temporaneo, il job potrebbe avere esito positivo quando viene ritentato. Per ulteriori informazioni sull'impostazione dei valori per i nuovi tentativi in Cloud Scheduler, consulta Creare un job.

Esamina le violazioni degli obiettivi della pipeline

Le seguenti sezioni descrivono come esaminare le pipeline che non soddisfano gli obiettivi di rendimento.

Pipeline batch ricorrenti

Per un'analisi iniziale dell'integrità della pipeline, nella pagina Informazioni sulla pipeline della console Google Cloud, utilizza i grafici Stato del singolo job e Tempo del thread per passaggio. Questi grafici si trovano nel riquadro di stato della pipeline.

Esempio di indagine:

1. Hai una pipeline batch ricorrente che viene eseguita ogni ora 3 minuti dopo l'ora. In genere, ogni job viene eseguito per circa 9 minuti. Hai un obiettivo per il completamento di tutti i job in meno di 10 minuti.

2. Il grafico dello stato del job mostra che un job è stato eseguito per più di 10 minuti.

3. Nella tabella della cronologia Aggiornamento/esecuzione, trova il job eseguito durante l'ora di interesse. Fai clic sulla pagina dei dettagli del job Dataflow. In questa pagina, individua la fase di esecuzione più lunga e cerca nei log eventuali errori per determinare la causa del ritardo.

Pipeline in modalità flusso

Per un'analisi iniziale dell'integrità della pipeline, utilizza il grafico di aggiornamento dei dati della pagina Dettagli pipeline della scheda Informazioni della pipeline. Questo grafico si trova nel riquadro dello stato della pipeline.

Esempio di indagine:

1. Hai una pipeline in modalità flusso che normalmente produce un output con un aggiornamento dei dati di 20 secondi.

2. Definisci l'obiettivo di avere una garanzia di aggiornamento dei dati di 30 secondi. Quando esamini il grafico, noti che tra le 09:00 e le 10:00 l'aggiornamento dei dati è salito a quasi 40 secondi.

   

3. Passa alla scheda Metriche pipeline, quindi visualizza i grafici Utilizzo della CPU e Utilizzo della memoria per ulteriori analisi.

Errore: l'ID pipeline esiste già all'interno del progetto

Se provi a creare una nuova pipeline con un nome già esistente nel tuo progetto, riceverai questo messaggio di errore: Pipeline Id already exist within the project. Per evitare questo problema, scegli sempre nomi univoci per le pipeline.

Nota: nonostante l'errore, il job Cloud Scheduler associato potrebbe comunque essere creato correttamente. Devi accedere alla console di Cloud Scheduler e verificare se il job esiste. Se necessario, annulla il job.