Questa pagina descrive le caratteristiche di rendimento dei job di streaming Dataflow che leggono da Pub/Sub e scrivono in BigQuery. Fornisce i risultati dei test benchmark per due tipi di pipeline di streaming:
Solo mappa (trasformazione per messaggio): pipeline che eseguono trasformazioni per messaggio, senza tenere traccia dello stato o raggruppare gli elementi nel flusso. Esempi: ETL, convalida dei campi e mappatura dello schema.
Aggregazione con finestra (
GroupByKey): pipeline che eseguono operazioni con stato e raggruppano i dati in base a una chiave e a una finestra temporale. Esempi: conteggio degli eventi, calcolo delle somme e raccolta dei record per una sessione utente.
La maggior parte dei carichi di lavoro per l'integrazione dei dati di streaming rientra in queste due categorie. Se la tua pipeline segue un pattern simile, puoi utilizzare questi benchmark per valutare il tuo job Dataflow rispetto a una configurazione di riferimento con un buon rendimento.
Metodologia di test
I benchmark sono stati condotti utilizzando le seguenti risorse:
Un argomento Pub/Sub di cui è stato eseguito il provisioning preliminare con un carico di input costante. I messaggi sono stati generati utilizzando il modello Generatore di dati di streaming.
- Frequenza dei messaggi: circa 1.000.000 di messaggi al secondo
- Carico di input: 1 GiB/s
- Formato dei messaggi: testo JSON generato in modo casuale con uno schema fisso
- Dimensioni dei messaggi: circa 1 KiB per messaggio
Una tabella BigQuery standard.
Pipeline di streaming Dataflow basate sul modello Da Pub/Sub a BigQuery. Queste pipeline eseguono l'analisi e la mappatura dello schema minime richieste. Non è stata utilizzata alcuna funzione definita dall'utente (UDF) personalizzata.
Dopo che la scalabilità orizzontale si è stabilizzata e la pipeline ha raggiunto lo stato stazionario, le pipeline sono state eseguite per circa un giorno, dopodiché i risultati sono stati raccolti e analizzati.
Pipeline Dataflow
Sono state testate due varianti di pipeline:
Pipeline solo mappa. Questa pipeline esegue una semplice mappatura e conversione dei messaggi JSON. Per questo test, il modello Da Pub/Sub a BigQuery è stato utilizzato senza modifiche.
- Semantica: la pipeline è stata testata utilizzando sia la modalità "exactly-once" sia la modalità "at-least-once". L'elaborazione "at-least-once" offre una velocità effettiva migliore. Tuttavia, deve essere utilizzata solo quando i record duplicati sono accettabili o il sink downstream gestisce la deduplicazione.
Pipeline di aggregazione con finestra. Questa pipeline raggruppa i messaggi in base a una chiave specifica in finestre di dimensioni fisse e scrive i record aggregati in BigQuery. Per questo test è stata utilizzata una pipeline Apache Beam personalizzata basata sul modello Da Pub/Sub a BigQuery.
Logica di aggregazione: per ogni finestra fissa e non sovrapposta di 1 minuto, i messaggi con la stessa chiave sono stati raccolti e scritti come un singolo record aggregato in BigQuery. Questo tipo di aggregazione è di uso comune nell'elaborazione dei log per combinare eventi correlati, ad esempio l'attività di un utente, in un singolo record per l'analisi downstream.
Parallelismo delle chiavi: il benchmark ha utilizzato 1.000.000 di chiavi distribuite in modo uniforme.
Semantica: la pipeline è stata testata utilizzando la modalità "exactly-once". Le aggregazioni richiedono una semantica "exactly-once" per garantire la correttezza ed evitare il doppio conteggio all'interno di un gruppo e di una finestra.
Configurazione job
La tabella seguente mostra come sono stati configurati i job Dataflow.
| Impostazione | Solo mappa, "exactly-once" | Solo mappa, "at-least-once" | Aggregazione con finestra, "exactly-once" |
|---|---|---|---|
| Tipo di macchina worker | n1-standard-2 |
n1-standard-2 |
n1-standard-2 |
| vCPU macchina worker | 2 | 2 | 2 |
| RAM macchina worker | 7,5 GiB | 7,5 GiB | 7,5 GiB |
| Persistent Disk macchina worker | Disco permanente standard (HDD), 30 GB | Disco permanente standard (HDD), 30 GB | Disco permanente standard (HDD), 30 GB |
| Worker iniziali | 70 | 30 | 180 |
| Numero massimo di worker | 100 | 100 | 250 |
| Streaming Engine | Yes | Sì | Yes |
| Scalabilità automatica orizzontale | Yes | Sì | Yes |
| Modello di fatturazione | Fatturazione basata sulle risorse | Fatturazione basata sulle risorse | Fatturazione basata sulle risorse |
| API Storage Write (gRPC) abilitata? | Yes | Sì | Yes |
| Stream API Storage Write (gRPC) | 200 | Non applicabile | 500 |
| Frequenza di attivazione API Storage Write (gRPC) | 5 secondi | Non applicabile | 5 secondi |
L'API BigQuery Storage Write (gRPC) è consigliata per le pipeline di streaming. Quando utilizzi la modalità "exactly-once" con l'API Storage Write (gRPC), puoi modificare le seguenti impostazioni:
Numero di stream di scrittura. Per garantire un parallelismo delle chiavi sufficiente nella fase di scrittura, imposta il numero di stream dell'API Storage Write (gRPC) su un valore maggiore del numero di CPU worker, mantenendo al contempo un livello ragionevole di velocità effettiva dello stream di scrittura BigQuery .
Frequenza di attivazione. Un valore di secondi a una cifra è adatto per le pipeline a velocità effettiva elevata.
Per ulteriori informazioni, consulta Scrivere da Dataflow a BigQuery.
Risultati benchmark
Questa sezione descrive i risultati dei test benchmark.
Velocità effettiva e utilizzo delle risorse
La tabella seguente mostra i risultati dei test per la velocità effettiva della pipeline e l'utilizzo delle risorse.
| Risultato | Solo mappa, "exactly-once" | Solo mappa, "at-least-once" | Aggregazione con finestra, "exactly-once" |
|---|---|---|---|
| Velocità effettiva di input per worker | Media: 17 MBps, n=3 | Media: 21 MBps, n=3 | Media: 6 MBps, n=3 |
| Utilizzo medio della CPU in tutti i worker | Media: 65%, n=3 | Media: 69%, n=3 | Media: 80%, n=3 |
| Numero di nodi worker | Media: 57, n=3 | Media: 48, n=3 | Media: 169, n=3 |
| Unità di calcolo Streaming Engine all'ora | Media: 125, n=3 | Media: 46, n=3 | Media: 354, n=3 |
L'algoritmo di scalabilità automatica può influire sul livello di utilizzo della CPU target. Per ottenere un utilizzo della CPU target più alto o più basso, puoi impostare l' intervallo di scalabilità automatica o il suggerimento per l'utilizzo dei worker. Target di utilizzo più elevati possono comportare costi inferiori, ma anche una latenza di coda peggiore, soprattutto per carichi variabili.
Per una pipeline di aggregazione con finestra, il tipo di aggregazione, le dimensioni della finestra e il parallelismo delle chiavi possono avere un impatto significativo sull'utilizzo delle risorse.
Latenza
La tabella seguente mostra i risultati benchmark per la latenza della pipeline.
| Latenza end-to-end totale della fase | Solo mappa, "exactly-once" | Solo mappa, "at-least-once" | Aggregazione con finestra, "exactly-once" |
|---|---|---|---|
| P50 | Media: 800 ms, n=3 | Media: 160 ms, n=3 | Media: 3400 ms, n=3 |
| P95 | Media: 2000 ms, n=3 | Media: 250 ms, n=3 | Media: 13.000 ms, n=3 |
| P99 | Media: 2800 ms, n=3 | Media: 410 ms, n=3 | Media: 25.000 ms, n=3 |
I test hanno misurato la latenza end-to-end per fase
(la job/streaming_engine/stage_end_to_end_latencies
metrica) in tre esecuzioni di test a lunga esecuzione. Questa metrica misura il tempo trascorso da Streaming Engine in ogni fase della pipeline. Include tutti i passaggi interni della pipeline, ad esempio:
- Shuffle e accodamento dei messaggi per l'elaborazione
- Il tempo di elaborazione effettivo, ad esempio la conversione dei messaggi in oggetti riga
- Scrittura dello stato persistente, nonché tempo trascorso in coda per scrivere lo stato persistente
Un'altra metrica di latenza è l'aggiornamento dei dati. Tuttavia, l'aggiornamento dei dati è influenzato da fattori quali la finestra definita dall'utente e i ritardi upstream nell'origine. La latenza del sistema fornisce una baseline più oggettiva per l'efficienza e l'integrità dell'elaborazione interna di una pipeline sotto carico.
I dati sono stati misurati per circa un giorno per esecuzione, con i periodi di avvio iniziali ignorati per riflettere un rendimento stabile e stazionario. I risultati mostrano due fattori che introducono una latenza aggiuntiva:
Modalità "exactly-once". Per ottenere una semantica "exactly-once", sono necessari shuffle deterministici e ricerche di stato persistenti per la deduplicazione. La modalità "at-least-once" è notevolmente più veloce perché ignora questi passaggi.
Aggregazione con finestra. I messaggi devono essere sottoposti a shuffle, memorizzati nel buffer e scritti nello stato persistente prima della chiusura della finestra, il che aumenta la latenza end-to-end.
I benchmark mostrati qui rappresentano una baseline. La latenza è molto sensibile alla complessità della pipeline. Le UDF personalizzate, le trasformazioni aggiuntive e la logica di finestra complessa possono aumentare la latenza. Le aggregazioni semplici e altamente riduttive, come la somma e il conteggio, tendono a generare una latenza inferiore rispetto alle operazioni con stato elevato, come la raccolta di elementi in un elenco.
Stima i costi
Puoi stimare il costo di base della tua pipeline comparabile con la fatturazione basata sulle risorse utilizzando il Google Cloud calcolatore prezzi, come segue:
- Apri il calcolatore prezzi.
- Fai clic su Aggiungi alla stima.
- Seleziona Dataflow.
- Per Tipo di servizio, seleziona "Dataflow Classic".
- Seleziona Impostazioni avanzate per visualizzare l'insieme completo di opzioni.
- Scegli la località in cui viene eseguito il job.
- Per Tipo di prestazione, seleziona "Streaming".
- Seleziona Abilita Streaming Engine.
- Inserisci le informazioni relative alle ore di esecuzione del job, ai nodi worker, alle macchine worker e allo spazio di archiviazione su Persistent Disk.
- Inserisci il numero stimato di unità di calcolo Streaming Engine.
L'utilizzo delle risorse e il costo aumentano in modo approssimativamente lineare con la velocità effettiva di input, anche se per i job di piccole dimensioni con solo pochi worker, il costo totale è dominato dai costi fissi. Come punto di partenza, puoi estrapolare il numero di nodi worker e il consumo di risorse dai risultati benchmark.
Supponiamo, ad esempio, di eseguire una pipeline solo mappa in modalità "exactly-once", con una velocità di dati di input di 100 MiB/s. In base ai risultati benchmark per una pipeline da 1 GiB/s, puoi stimare i requisiti delle risorse come segue:
- Fattore di scalabilità: (100 MiB/s) / (1 GiB/s) = 0,1
- Nodi worker previsti: 57 worker × 0,1 = 5,7 worker
- Numero previsto di unità di calcolo Streaming Engine all'ora: 125 × 0,1 = 12,5 unità all'ora
Questo valore deve essere utilizzato solo come stima iniziale. La velocità effettiva e il costo effettivi possono variare in modo significativo in base a fattori quali il tipo di macchina, la distribuzione delle dimensioni dei messaggi, il codice utente, il tipo di aggregazione, il parallelismo delle chiavi e le dimensioni della finestra. Per ulteriori informazioni, consulta Best practice per l'ottimizzazione dei costi di Dataflow.
Esegui una pipeline di test
Questa sezione mostra i
gcloud dataflow flex-template run
comandi utilizzati per eseguire la pipeline solo mappa.
Modalità "exactly-once"
gcloud dataflow flex-template run JOB_ID \
--template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/PubSub_to_BigQuery_Flex \
--enable-streaming-engine \
--num-workers 70 \
--max-workers 100 \
--parameters \
inputSubscription=projects/PROJECT_IDsubscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
useStorageWriteApi=true,\
numStorageWriteApiStreams=200 \
storageWriteApiTriggeringFrequencySec=5
Modalità "at-least-once"
gcloud dataflow flex-template run JOB_ID \
--template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/PubSub_to_BigQuery_Flex \
--enable-streaming-engine \
--num-workers 30 \
--max-workers 100 \
--parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
useStorageWriteApi=true \
--additional-experiments streaming_mode_at_least_once
Sostituisci quanto segue:
JOB_ID: l'ID job DataflowPROJECT_ID: l'ID progettoSUBSCRIPTION_NAME: il nome della sottoscrizione Pub/SubDATASET: il nome del set di dati BigQueryTABLE_NAME: il nome della tabella BigQuery
Genera dati di test
Per generare dati di test, utilizza il seguente comando per eseguire il modello Generatore di dati di streaming:
gcloud dataflow flex-template run JOB_ID \
--template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/Streaming_Data_Generator \
--num-workers 70 \
--max-workers 100 \
--parameters \
topic=projects/PROJECT_ID/topics/TOPIC_NAME,\
qps=1000000,\
maxNumWorkers=100,\
schemaLocation=SCHEMA_LOCATION
Sostituisci quanto segue:
JOB_ID: l'ID job DataflowPROJECT_ID: l'ID progettoTOPIC_NAME: il nome dell'argomento Pub/SubSCHEMA_LOCATION: il percorso di un file di schema in Cloud Storage
Il modello Generatore di dati di streaming utilizza un file JSON Data Generator per definire lo schema dei messaggi. I test benchmark hanno utilizzato uno schema di messaggi simile al seguente:
{ "logStreamId": "{{integer(1000001,2000000)}}", "message": "{{alphaNumeric(962)}}" }