Utilizzare 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 di Dataflow per le seguenti attività:

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

Per la documentazione dell'API, consulta la pagina 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 la capacità aggregata all'uso e al consumo di risorse di una pipeline.
• Visualizza l'aggiornamento dei dati di una pipeline in streaming. Questa metrica, che si evolve nell'intervallo di tempo, può essere collegato a un avviso che ti avvisa quando l'aggiornamento scende al di sotto di una dell'obiettivo specificato.
• Utilizza i grafici delle metriche della pipeline per confrontare i job della pipeline batch e trovare anomalie.

Limitazioni

• Disponibilità regionale: puoi creare pipeline di dati in regioni disponibili per 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 a livello di organizzazione e, se lo fai, ogni organizzazione può avere per impostazione predefinita nella maggior parte delle pipeline 2500.

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

• Etichette: non puoi utilizzare etichette definite dall'utente per etichettare le pipeline di dati di Dataflow. Tuttavia, quando utilizzi additionalUserLabels questi valori vengono trasmessi al tuo job Dataflow. Per ulteriori informazioni su come le etichette si applicano a singole Job Dataflow, consulta Opzioni pipeline.

Tipi di pipeline di dati

Dataflow ha due tipi di pipeline di dati: flusso e batch. Entrambi i tipi di pipeline di esecuzione dei job definiti in Dataflow modelli.

Pipeline di dati in modalità flusso

Una pipeline di dati in modalità flusso esegue un flusso di dati Dataflow un job esistente subito dopo la sua creazione.

Pipeline di dati in batch

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

Nota: ogni nome di job batch di Dataflow creato da una pipeline di dati batch utilizza il seguente pattern di denominazione: <pipeline_id>-mp--<timestamp>-<random int>. Il valore di timestamp ha una granularità in secondi. La stringa -mp– è riservata per indicare che Il job batch di Dataflow viene creato da una pipeline di dati batch. pipeline_id viene troncato dopo 27 caratteri.

Pipeline batch incrementali

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

• È possibile usare segnaposto per anno, mese, data, ora, minuto e secondo. devono rispettare le 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 parametro, viene valutato come "gs://bucket/Y", perché "Y" senza un "%" precedente non viene mappato al formato strftime().

A ogni tempo di esecuzione della pipeline batch pianificata, parte del percorso di input viene valutata nello stato time-shifted) per data/ora. I valori di data 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 acquisito per l'elaborazione dalla pipeline batch all'ora pianificata.

• Esempio: una pipeline batch è pianificata per ripetersi all'inizio di ogni ora PST. Se parametri 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 gs://bucket-name/2021-04-15/prefix-18_00.csv.

Utilizza i parametri di spostamento temporale

Puoi utilizzare i parametri di sfasamento dell'ora di + o - minuti o ore. Per supportare la corrispondenza di un percorso di input con una data e ora valutata spostata prima o dopo la data e ora corrente della pianificazione della pipeline,chiudere questi parametri tra parentesi graffe. Utilizza il formato {[+|-][0-9]+[m|h]}. La pipeline batch continua a ripetersi all'ora pianificata, ma il percorso di input viene valutato con in offset temporale.

• Esempio: una pipeline batch viene pianificata per ripetersi all'inizio di ogni ora PST. Se parametri 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 gs://bucket-name/2021-04-15/prefix-16_00.csv.

Ruoli della pipeline di dati

Affinché le operazioni della pipeline di dati Dataflow vadano a buon fine, è necessaria ruoli IAM necessari, come segue:

1. Per eseguire le operazioni, devi disporre del 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 della pipeline di dati (questo ruolo può essere abilitato utilizzando l'API)

2. L'account di servizio utilizzato da Cloud Scheduler deve avere roles/iam.serviceAccountUser ruolo, Indica se l'account di servizio è specificato dall'utente o come Compute Engine predefinito l'account di servizio. Per saperne di più, consulta Ruoli della pipeline di dati.

3. Devi poter agire come account di servizio utilizzato da Cloud Scheduler e a Dataflow concedendo roles/iam.serviceAccountUser del ruolo per l'account. Se non selezioni un account di servizio per Cloud Scheduler e Dataflow, viene utilizzato il service account Compute Engine predefinito.

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 autorizzazioni mancanti per gli account di servizio per Cloud Scheduler o Dataflow. Verifica e correggi queste autorizzazioni per risolvere i problemi della 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 a Dataflow nella console Google Cloud, si apre una pagina di configurazione. Attiva il le API elencate per creare pipeline di dati.

Importare un job

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

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

   Vai a Job

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

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

4. Per un job batch, nella sezione Pianifica la pipeline, fornisci una pianificazione della ricorrenza. Fornire un indirizzo di account email per Cloud Scheduler, che viene utilizzato per pianificare le esecuzioni batch, è facoltativo. In caso contrario specificato, account di servizio Compute Engine predefinito .

Crea una pipeline di dati

1. Nella console Google Cloud, vai a Dataflow Pipeline di dati.

   Vai a Pipeline di dati

2. Seleziona + Crea pipeline di dati.

3. Nella pagina Crea pipeline da modello, fornisci un nome pipeline e compila l'altro modello di selezione e parametri.

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

Crea una pipeline di dati in batch

Per creare questa pipeline di dati in batch di esempio, devi dispongono dell'accesso alle seguenti risorse nel tuo progetto:

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

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

1. Crea i seguenti file sul tuo disco 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 macro 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 in 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 a Bucket di Cloud Storage .

   Vai a Bucket

4. Per creare una cartella tmp nel tuo 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 a Dataflow Pipeline di dati.

   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. Pertanto, Il bucket Cloud Storage e la tabella BigQuery devono trovarsi nella stessa regione.

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

      Nota: non selezionare la pipeline di inserimento flussi con lo stesso nome in Elaborare i dati in modo continuo (flusso).

   4. Per Pianifica la pipeline, seleziona una pianificazione, ad esempio Orari al minuto 25. nel tuo fuso orario. Puoi modificare la pianificazione dopo aver inviato la pipeline. Fornire un indirizzo email per Cloud Scheduler, che viene utilizzato per pianificare le esecuzioni collettive, è facoltativo. In caso contrario specificato, account di servizio Compute Engine predefinito .

   5. In Parametri obbligatori, inserisci quanto segue:

      1. Per il percorso della funzione JavaScript definita dall'utente in Cloud Storage:

         gs://BUCKET_ID/text_to_bigquery/split_csv_3cols.js
         

      2. Per il percorso JSON:

         BUCKET_ID/text_to_bigquery/bq_three_column_table.json
         

      3. Per Nome della funzione JavaScript definita dall'utente: transform
      4. Per 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 su pipeline e modello e visualizzale la cronologia attuale e quella precedente dalla pagina Dettagli pipeline.

   

Puoi modificare la pianificazione della pipeline di dati dal riquadro Informazioni pipeline nella Pagina Dettagli pipeline.

Puoi anche eseguire una pipeline batch su richiesta utilizzando il pulsante Esegui nella console delle pipeline Dataflow.

Crea una pipeline di dati in modalità flusso di esempio

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

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

Risoluzione dei problemi

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

Impossibile avviare il job della pipeline di dati

Quando utilizzi le pipeline di dati per creare una pianificazione di job ricorrenti, Il job Dataflow potrebbe non essere avviato e viene visualizzato un errore di stato 503 in dei log di Cloud Scheduler.

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

Per risolvere il problema, configura Cloud Scheduler per riprovare il job. Poiché il problema è temporaneo. Se il job viene ritentato, potrebbe riuscire. Per per ulteriori informazioni sull'impostazione dei valori per i nuovi tentativi in Cloud Scheduler, consulta Crea un job.

Esaminare le violazioni degli obiettivi della pipeline

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

Pipeline batch ricorrenti

Per un'analisi iniziale dell'integrità della pipeline, Nella pagina Informazioni pipeline della console Google Cloud, utilizza il metodo Grafici dello stato dei singoli job e di Tempo di thread per passaggio. Questi grafici si trovano nel riquadro dello stato della pipeline.

Esempio di indagine:

1. Hai una pipeline batch ricorrente che viene eseguita ogni ora dopo 3 minuti all'ora. Normalmente, ogni job viene eseguito per circa 9 minuti. Hai un obiettivo che tutti i job vengano completati 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, individua il job eseguito durante ora di interesse. Fai clic per arrivare alla pagina dei dettagli del job Dataflow. In questa pagina, trova la fase più lunga in esecuzione e cerca nei log la presenza di possibili errori per determinare la causa del ritardo.

Pipeline in modalità flusso

Per un'analisi iniziale dell'integrità della pipeline, nella pagina Dettagli pipeline, nella scheda Informazioni sulla pipeline, utilizza il grafico sull'aggiornamento dei dati. Questo grafico si trova nel riquadro Stato pipeline.

Indagine di esempio:

1. Hai una pipeline in modalità flusso che normalmente produce un output con 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 dell'aggiornamento dei dati, noti che tra le 9 e le 10:00, dell'aggiornamento dei dati è balzato 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à nel progetto

Se provi a creare una nuova pipeline con un nome che esiste già progetto, viene visualizzato 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 essere comunque creato correttamente. Devi accedere alla console di Cloud Scheduler e verificare se il job esiste. Se necessario, annulla il job.