Panoramica del pushdown delle trasformazioni

Per migliorare le prestazioni delle pipeline di dati, puoi eseguire il push di alcune operazioni di trasformazione in BigQuery anziché in Apache Spark. Pushdown di trasformazione si riferisce a un'impostazione che consente di inviare un'operazione in una pipeline di dati di Cloud Data Fusion a BigQuery come motore di esecuzione. Di conseguenza, l'operazione e i relativi dati vengono trasferiti a BigQuery e l'operazione viene eseguita lì.

Il pushdown della trasformazione migliora le prestazioni delle pipeline con più operazioni JOIN complesse o altre trasformazioni supportate. L'esecuzione di alcune trasformazioni in BigQuery potrebbe essere più veloce rispetto all'esecuzione in Spark.

Le trasformazioni non supportate e tutte le trasformazioni di anteprima vengono eseguite in Spark.

Trasformazioni supportate

Il pushdown della trasformazione è disponibile in Cloud Data Fusion versione 6.5.0 e successive, ma alcune delle seguenti trasformazioni sono supportate solo nelle versioni successive.

Nota: quando visualizzi l'anteprima dei dati in Cloud Data Fusion, tutte le trasformazioni vengono eseguite in Spark e non in BigQuery.

JOIN operazioni

• Il pushdown della trasformazione è disponibile per le operazioni JOIN in Cloud Data Fusion versione 6.5.0 e successive.

• Sono supportate le operazioni di base (sui tasti) e avanzate JOIN.

• Per l'esecuzione in BigQuery, i join devono avere esattamente due fasi di input.

• I join configurati per caricare uno o più input in memoria vengono eseguiti in Spark anziché in BigQuery, tranne nei seguenti casi:

  • Se uno degli input dell'unione è già stato eseguito il push.
  • Se hai configurato l'unione da eseguire in SQL Engine (vedi l'opzione Fasi per forzare l'esecuzione).

Sink BigQuery

Il pushdown di trasformazione è disponibile per il sink BigQuery in Cloud Data Fusion versione 6.7.0 e successive.

Quando il sink BigQuery segue una fase eseguita in BigQuery, l'operazione che scrive i record in BigQuery viene eseguita direttamente in BigQuery.

Per migliorare il rendimento con questo sink, devi disporre di quanto segue:

• Il account di servizio deve disporre dell'autorizzazione per creare e aggiornare le tabelle nel set di dati utilizzato dal sink BigQuery.
• I set di dati utilizzati per il pushdown della trasformazione e il sink BigQuery devono essere archiviati nella stessa posizione.
• L'operazione deve essere una delle seguenti:
  • Insert (l'opzione Truncate Table non è supportata)
  • Update
  • Upsert

Nota: senza questi requisiti, l'operazione di sink utilizza il flusso di lavoro esistente e non esegue una copia diretta.

GROUP BY aggregazioni

Il pushdown della trasformazione è disponibile per le aggregazioni GROUP BY in Cloud Data Fusion versione 6.7.0 e successive.

Le aggregazioni GROUP BY in BigQuery sono disponibili per le seguenti operazioni:

• Avg
• Collect List (i valori nulli vengono rimossi dall'array di output)
• Collect Set (i valori nulli vengono rimossi dall'array di output)
• Concat
• Concat Distinct
• Count
• Count Distinct
• Count Nulls
• Logical And
• Logical Or
• Max
• Min
• Standard Deviation
• Sum
• Sum of Squares
• Corrected Sum of Squares
• Variance
• Shortest String
• Longest String

Nota: le aggregazioni condizionali, che consentono di utilizzare un'espressione di filtro prima di aggregare i record, non sono supportate. Se viene utilizzata un'aggregazione condizionale, l'operazione GROUP BY viene eseguita in Spark.

Gli aggregatori GROUP BY vengono eseguiti in BigQuery nei seguenti casi:

• Segue una fase che è già stata spostata in basso.
• L'hai configurato per l'esecuzione in SQL Engine (vedi l'opzione Fasi per forzare l'esecuzione).

Deduplica le aggregazioni

Il pushdown della trasformazione è disponibile per le aggregazioni di deduplicazione in Cloud Data Fusion versione 6.7.0 e successive per le seguenti operazioni:

• Nessuna operazione di filtro specificata
• ANY (un valore non nullo per il campo desiderato)
• MIN (il valore minimo per il campo specificato)
• MAX (il valore massimo per il campo specificato)

Le seguenti operazioni non sono supportate:

• FIRST
• LAST

Le aggregazioni di deduplicazione vengono eseguite nel motore SQL nei seguenti casi:

• Segue una fase che è già stata spostata in basso.
• L'hai configurato per l'esecuzione in SQL Engine (vedi l'opzione Fasi per forzare l'esecuzione).

Pushdown dell'origine BigQuery

Il pushdown dell'origine BigQuery è disponibile nelle versioni 6.8.0 e successive di Cloud Data Fusion.

Quando un'origine BigQuery segue una fase compatibile con il pushdown di BigQuery, la pipeline può eseguire tutte le fasi compatibili all'interno di BigQuery.

Cloud Data Fusion copia i record necessari per eseguire la pipeline in BigQuery.

Quando utilizzi il pushdown dell'origine BigQuery, le proprietà di partizionamento e clustering della tabella vengono mantenute, il che ti consente di utilizzarle per ottimizzare ulteriormente le operazioni, ad esempio i join.

Requisiti aggiuntivi

Per utilizzare il pushdown dell'origine BigQuery, devono essere soddisfatti i seguenti requisiti:

• Il account di servizio configurato per il pushdown della trasformazione BigQuery deve disporre delle autorizzazioni per leggere le tabelle nel set di dati dell'origine BigQuery.

• I set di dati utilizzati nell'origine BigQuery e il set di dati configurato per il pushdown della trasformazione devono essere archiviati nella stessa posizione.

Aggregazioni finestra

Il pushdown della trasformazione è disponibile per le aggregazioni di finestre in Cloud Data Fusion versione 6.9 e successive. Le aggregazioni delle finestre in BigQuery sono supportate per le seguenti operazioni:

• Rank
• Dense Rank
• Percent Rank
• N tile
• Row Number
• Median
• Continuous Percentile
• Lead
• Lag
• First
• Last
• Cumulative distribution
• Accumulate

Le aggregazioni delle finestre vengono eseguite in BigQuery nei seguenti casi:

• Segue una fase che è già stata spostata in basso.
• L'hai configurato per l'esecuzione in SQL Engine (vedi l'opzione Fasi per forzare il pushdown).

Pushdown del filtro Wrangler

Il pushdown del filtro Wrangler è disponibile in Cloud Data Fusion versione 6.9 e successive.

Quando utilizzi il plug-in Wrangler, puoi eseguire il push dei filtri, noti come operazioni Precondition, in BigQuery anziché in Spark.

Il push-down dei filtri è supportato solo con la modalità SQL per Preconditions, rilasciata anch'essa nella versione 6.9. In questa modalità, il plug-in accetta un'espressione di precondizione in SQL standard ANSI.

Se la modalità SQL viene utilizzata per le precondizioni, le direttive e le direttive definite dall'utente sono disattivate per il plug-in Wrangler, in quanto non sono supportate con le precondizioni in modalità SQL.

La modalità SQL per le precondizioni non è supportata per i plug-in Wrangler con più input quando il pushdown della trasformazione è attivato. Se utilizzata con più input, questa fase di Wrangler con condizioni di filtro SQL viene eseguita in Spark.

I filtri vengono eseguiti in BigQuery nei seguenti casi:

• Segue una fase che è già stata spostata in basso.
• L'hai configurato per l'esecuzione in SQL Engine (vedi l'opzione Fasi per forzare il pushdown).

Metriche

Per ulteriori informazioni sulle metriche fornite da Cloud Data Fusion per la parte della pipeline eseguita in BigQuery, consulta Metriche della pipeline di pushdown di BigQuery.

Quando utilizzare il pushdown della trasformazione

L'esecuzione delle trasformazioni in BigQuery prevede quanto segue:

1. Scrittura dei record in BigQuery per le fasi supportate della pipeline.
2. Esecuzione delle fasi supportate in BigQuery.
3. Lettura dei record da BigQuery dopo l'esecuzione delle trasformazioni supportate, a meno che non siano seguite da un sink BigQuery.

A seconda delle dimensioni dei set di dati, potrebbe esserci un overhead di rete considerevole, che può avere un impatto negativo sul tempo di esecuzione complessivo della pipeline quando il pushdown della trasformazione è abilitato.

A causa dell'overhead di rete, consigliamo il pushdown della trasformazione nei seguenti casi:

• Più operazioni supportate vengono eseguite in sequenza (senza passaggi tra le fasi).
• I miglioramenti delle prestazioni ottenuti con BigQuery che esegue le trasformazioni, rispetto a Spark, superano la latenza del trasferimento dei dati in BigQuery e, possibilmente, fuori da BigQuery.

Come funziona

Quando esegui una pipeline che utilizza il pushdown di trasformazione, Cloud Data Fusion esegue le fasi di trasformazione supportate in BigQuery. Tutte le altre fasi della pipeline vengono eseguite in Spark.

Durante l'esecuzione delle trasformazioni:

1. Cloud Data Fusion carica i set di dati di input in BigQuery scrivendo i record in Cloud Storage ed eseguendo un job di caricamento BigQuery.

2. Le operazioni JOIN e le trasformazioni supportate vengono quindi eseguite come job BigQuery utilizzando istruzioni SQL.

3. Se è necessaria un'ulteriore elaborazione dopo l'esecuzione dei job, i record possono essere esportati da BigQuery a Spark. Tuttavia, se l'opzione Tenta la copia diretta nei sink BigQuery è abilitata e il sink BigQuery segue una fase eseguita in BigQuery, i record vengono scritti direttamente nella tabella del sink BigQuery di destinazione.

Il seguente diagramma mostra come il pushdown della trasformazione esegue le trasformazioni supportate in BigQuery anziché in Spark.

Best practice

Regola le dimensioni del cluster e dell'esecutore

Per ottimizzare la gestione delle risorse nella pipeline:

• Utilizza il giusto numero di worker (nodi) del cluster per un workload. In altre parole, ottieni il massimo dal cluster del servizio gestito per Apache Spark di cui è stato eseguito il provisioning utilizzando completamente la CPU e la memoria disponibili per la tua istanza, beneficiando al contempo della velocità di esecuzione di BigQuery per i job di grandi dimensioni.

• Migliora il parallelismo nelle pipeline utilizzando cluster con scalabilità automatica.

• Modifica le configurazioni delle risorse nelle fasi della pipeline in cui i record vengono inviati o recuperati da BigQuery durante l'esecuzione della pipeline.

Consigliato: prova ad aumentare il numero di core della CPU per le risorse dell'executor (fino al numero di core della CPU utilizzati dal nodo worker). Gli executor ottimizzano l'utilizzo della CPU durante i passaggi di serializzazione e deserializzazione quando i dati entrano e escono da BigQuery. Per saperne di più, consulta Dimensionamento del cluster.

Un vantaggio dell'esecuzione delle trasformazioni in BigQuery è che le pipeline possono essere eseguite su cluster Managed Service for Apache Spark più piccoli. Se i join sono le operazioni che richiedono più risorse nella pipeline, puoi provare con dimensioni del cluster più piccole, poiché le operazioni JOIN pesanti vengono ora eseguite in BigQuery, il che ti consente di ridurre potenzialmente i costi di calcolo complessivi.

Recuperare i dati più rapidamente con l'API BigQuery Storage Read

Dopo che BigQuery esegue le trasformazioni, la pipeline potrebbe avere fasi aggiuntive da eseguire in Spark. In Cloud Data Fusion versione 6.7.0 e successive, il pushdown di trasformazione supporta l'API BigQuery Storage Read, che migliora la latenza e velocizza le operazioni di lettura in Spark. Può ridurre il tempo di esecuzione complessivo della pipeline.

L'API legge i record in parallelo, quindi ti consigliamo di regolare le dimensioni dell'executor di conseguenza. Se in BigQuery vengono eseguite operazioni che richiedono molte risorse, riduci l'allocazione della memoria per gli executor per migliorare il parallelismo durante l'esecuzione della pipeline (vedi Modificare le dimensioni del cluster e dell'executor).

L'API BigQuery Storage di lettura è disabilitata per impostazione predefinita. Puoi abilitarlo negli ambienti di esecuzione in cui è installato Scala 2.12 (inclusi Managed Service for Apache Spark 2.0 e Managed Service for Apache Spark 1.5).

Nota: l'API BigQuery Storage di lettura comporta costi aggiuntivi (vedi Prezzi).

Considera le dimensioni del set di dati

Considera le dimensioni dei set di dati nelle operazioni JOIN. Per le operazioni JOIN che generano un numero elevato di record di output, ad esempio un'operazione simile a un prodotto incrociato JOIN, la dimensione del set di dati risultante potrebbe essere di ordini di grandezza superiore a quella del set di dati di input. Inoltre, considera l'overhead del recupero di questi record in Spark quando si verifica un'ulteriore elaborazione Spark per questi record, ad esempio una trasformazione o un sink, nel contesto del rendimento complessivo della pipeline.

Mitigare i dati distorti

Le operazioni JOIN per i dati con forte asimmetria potrebbero causare il superamento dei limiti di utilizzo delle risorse da parte del job BigQuery, il che comporta l'esito negativo dell'operazione JOIN. Per evitare questo problema, vai alle impostazioni del plug-in Joiner e identifica l'input distorto nel campo Skewed Input Stage. In questo modo, Cloud Data Fusion organizza gli input in modo da ridurre il rischio che l'istruzione BigQuery superi i limiti.

Passaggi successivi

• Scopri come attivare il pushdown della trasformazione in Cloud Data Fusion.