Utilizza Datastream e Dataflow per trasmettere dati in BigQuery

Quando replichi i dati in BigQuery utilizzando la pipeline di streaming Datastream e Dataflow, puoi personalizzare la modalità di elaborazione e unione delle modifiche da parte di Dataflow. Il partizionamento delle tabelle di replica in base a chiavi personalizzate, la trasformazione degli eventi con le funzioni definite dall'utente (UDF) JavaScript e l'ottimizzazione della frequenza di unione ti aiutano a bilanciare l'aggiornamento dei dati con i costi di query e archiviazione.

Partiziona i set di dati di replica in base alle chiavi definite dall'utente

Il set di dati di staging in BigQuery viene partizionato automaticamente. Tuttavia, per impostazione predefinita, il set di dati di replica non viene partizionato perché le chiavi di partizionamento nelle tabelle di replica devono essere definite in base a una logica di business specifica, anziché essere applicate da Datastream e Dataflow.

Per ogni tabella nel set di dati di replica che deve essere partizionata:

  1. Arresta e svuota il job Dataflow

  2. Utilizza l'editor SQL in BigQuery per eseguire il seguente script SQL per ogni tabella nel set di dati di replica. In questo esempio, la tabella actor nel set di dati datastream_cdc ha una colonna last_update che vogliamo impostare come chiave di partizionamento. Eseguendo lo script, ricrei la tabella con la chiave di partizionamento corretta.

    create table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' partition by date(last_update)
as SELECT * FROM '[BigQuery_PROJECT_ID].datastream_cdc.actor'

drop table '[BigQuery_PROJECT_ID].datastream_cdc.actor'

alter table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' rename to 'actor'

  3. Utilizza il modello Datastream a BigQuery per ricreare un job Dataflow.

Esegui funzioni definite dall'utente per manipolare i dati degli eventi

Puoi utilizzare il modello Datastream a BigQuery per eseguire una funzione definita dall'utente JavaScript. Per farlo, innanzitutto, inserisci un file contenente la funzione in una posizione specifica all'interno di Cloud Storage. Quindi:

  • Utilizza il parametro javascriptTextTransformGcsPath nel modello per specificare la posizione del file in Cloud Storage che contiene la funzione definita dall'utente.
  • Utilizza il parametro javascriptTextTransformFunctionName per specificare il nome della funzione JavaScript che vuoi chiamare come funzione definita dall'utente.

Ad esempio, puoi eseguire una funzione definita dall'utente per conservare i record eliminati nelle tabelle del set di dati di replica in BigQuery. Questa procedura è nota come eliminazione temporanea.

Per farlo, crea una funzione che copia il valore della colonna _metadata_deleted in una nuova colonna denominata is_deleted, quindi reimposta il valore della colonna _metadata_deleted su false. In questo modo, il job Dataflow ignora gli eventi di eliminazione e conserva i record eliminati durante l'aggiornamento del set di dati di replica in BigQuery.

Ecco il codice campione per questa funzione definita dall'utente:

/**
* This function changes the behavior of the Datastream to
* BigQuery template to allow soft deletes.
* @param {string} messageString from DatastreamIO data
* @return {string} same as an input message with an added property
*/
function transform(messageString) {
   // messageString is a JSON object as a string
   var messageJson = JSON.parse(messageString);
    // Moving the deleted flag to a new column will cause the pipeline to soft delete data.
   messageJson['is_deleted'] = messageJson['_metadata_deleted'];
   messageJson['_metadata_deleted'] = false;
    return JSON.stringify(messageJson);
 }

Imposta la frequenza di unione

Utilizza il parametro mergeFrequencyMinutes del modello Datastream a BigQuery per impostare la frequenza di unione. Questo è il numero di minuti tra le unioni per una determinata tabella nel set di dati di replica in BigQuery. Durante il backfill dei dati storici, ti consigliamo di mantenere bassa la frequenza di unione (12 o 24 ore) per tenere sotto controllo i costi.

Ad esempio, se imposti il valore di questo parametro su 10 minuti, Dataflow eseguirà il job che utilizza il modello ogni 10 minuti. Tuttavia, la prima volta che il job viene eseguito, si verifica un ritardo di 5 minuti. In questo esempio, se il job viene eseguito alle 9:14, la prima unione avverrà alle 9:29 (10 minuti per l'unione e 5 minuti per il ritardo). La seconda unione avverrà alle 9:39 e tutte le unioni successive avverranno a intervalli di 10 minuti (9:49, 9:59, 10:09 e così via).

Se imposti la frequenza di unione su 60 minuti, il job verrà eseguito all'ora, dopo un ritardo di 5 minuti per l'esecuzione iniziale del job. Se l'esecuzione del job è pianificata per le 10:00, verrà eseguita alle 10:05 a causa del ritardo di 5 minuti. Tutte le unioni successive avverranno a intervalli di 60 minuti (11:05, 12:05, 13:05 e così via).

A causa del controllo dei costi o per altri motivi, potresti non essere in grado di eseguire un'unione con una frequenza che soddisfi le tue esigenze aziendali. Potresti non avere i dati più recenti. Per accedere ai dati più recenti, crea una visualizzazione sopra le tabelle dei set di dati di staging e di replica in BigQuery, in cui la visualizzazione simula l'unione. Questa visualizzazione viene creata come una tabella logica (sia per i set di dati di staging che per quelli di replica). Se la frequenza di unione è bassa e hai bisogno di un accesso più rapido ai dati, utilizza la visualizzazione.