Se devi caricare messaggi AVRO o JSON da un argomento di Google Cloud Managed Service per Apache Kafka in una tabella BigQuery, puoi farlo con un modello Dataflow. Questo documento ti aiuta a configurare una pipeline basata su questo modello utilizzando la Google Cloud console. Puoi anche utilizzare l'API REST o Google Cloud CLI per configurare questa pipeline.
Google Cloud Prodotti utilizzati
Il modello Dataflow da Kafka a BigQuery utilizza i seguenti prodotti a pagamento Google Cloud . Utilizza il calcolatore prezzi per generare una stima dei costi in base all'utilizzo previsto.
- BigQuery: BigQuery è il data warehouse serverless di Google Google Cloud. In questa soluzione, BigQuery funge da destinazione per i dati Kafka. L'API BigQuery Streaming Write viene utilizzata per scrivere i dati in BigQuery. Puoi quindi utilizzare BigQuery per analizzare questi dati con query SQL, creare modelli di machine learning con BigQuery ML, e potenziare le tue applicazioni di business intelligence.
- Dataflow: Dataflow è un servizio di elaborazione dei dati completamente gestito. Il modello Dataflow da Kafka a BigQuery utilizza Dataflow per creare una pipeline che legge i dati dall'argomento Kafka, esegue le trasformazioni necessarie e li scrive in BigQuery. Le funzionalità di scalabilità automatica e autoriparazione di Dataflow garantiscono che la pipeline venga eseguita in modo affidabile ed efficiente.
- Cloud Storage: sebbene non sia direttamente coinvolto nel flusso di dati principale, Cloud Storage può essere utilizzato per archiviare i file dello schema Avro se utilizzi la codifica binaria Avro per i tuoi messaggi Kafka. In questo modo, la pipeline Dataflow può interpretare correttamente la struttura dei dati.
Inoltre, la soluzione utilizza anche Google Cloud Managed Service per Apache Kafka.
- Google Cloud Managed Service per Apache Kafka: Un Google Cloud servizio che ti aiuta a eseguire Apache Kafka. Managed Service per Apache Kafka ti consente di concentrarti sulla creazione di sistemi basati su eventi e pipeline di flussi di dati, anziché sulla gestione dell'infrastruttura. Per ulteriori informazioni sui prezzi di Google Cloud Managed Service per Apache Kafka, consulta la guida ai prezzi.
Prima di iniziare
Prima di avviare il modello Dataflow da Kafka a BigQuery , assicurati di aver completato le seguenti operazioni:
Crea un cluster e un argomento Managed Service per Apache Kafka.
Un modo per creare un cluster e un argomento è seguire la guida rapida di Managed Service per Apache Kafka.
Se l'argomento contiene record Avro, per ulteriori requisiti di risorse, consulta Specificare il formato del messaggio.
Abilita le seguenti Google Cloud API:
Dataflow
BigQuery
Cloud Storage
gcloud services enable dataflow.googleapis.com bigquery.googleapis.com \ storage.googleapis.comCrea un set di dati e una tabella BigQuery. Assicurati che lo schema della tabella corrisponda allo schema dell'argomento di input Kafka.
Se utilizzi più schemi all'interno dello stesso argomento e scrivi in più tabelle, non devi creare una tabella prima di configurare la pipeline.
Per ulteriori informazioni su come creare un set di dati BigQuery e una tabella, consulta Creare un set di dati e Creare una tabella vuota con una definizione dello schema.
Concedi il ruolo di client Kafka gestito al account di servizio worker Dataflow
Per connettere il job Dataflow a Managed Service per Apache Kafka, devi concedere autorizzazioni specifiche al account di servizio worker Dataflow. Questo account di servizio è l'identità utilizzata per tutte le VM worker nel job Dataflow e tutte le richieste effettuate da queste VM utilizzano questo account.
Per consentire l'accesso alle risorse Kafka, devi concedere il ruolo roles/managedkafka.client al account di servizio worker Dataflow. Questo ruolo include l'autorizzazione managedkafka.clusters.connect necessaria per stabilire le connessioni.
Per ulteriori informazioni sul account di servizio worker, consulta Sicurezza e autorizzazioni per le pipeline su Google Cloud.
Per concedere il ruolo client Kafka gestito al account di servizio Dataflow, segui questi passaggi:
Console
- Nella Google Cloud console vai alla pagina IAM.
Vai a IAM - Verifica che il progetto sia impostato sul progetto consumer a cui il client Managed Service per Apache Kafka accederà.
- Fai clic su Concedi l'accesso.
- Nella nuova pagina, in Aggiungi entità, inserisci l'indirizzo email del account di servizio worker Dataflow che stai utilizzando.
- In Assegna i ruoli, seleziona il ruolo Client Kafka gestito.
- Fai clic su Salva.
Gcloud CLI
-
Nella Google Cloud console, attiva Cloud Shell.
Nella parte inferiore della Google Cloud console viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già inclusa e installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.
-
Esegui il
gcloud projects add-iam-policy-bindingcomando:gcloud projects add-iam-policy-binding PROJECT_ID \ --member serviceAccount:SERVICE_ACCOUNT_EMAIL \ --role roles/managedkafka.client
Sostituisci quanto segue:
-
PROJECT_ID è l'ID progetto.
-
SERVICE_ACCOUNT_EMAIL è l'indirizzo email del account di servizio worker Dataflow.
-
Avvia il modello Dataflow da Kafka a BigQuery
Puoi avviare il modello Dataflow da Kafka a BigQuery dalla pagina dei dettagli del cluster nella console.
-
Nella Google Cloud console vai alla pagina Cluster.
Vengono elencati i cluster creati in un progetto.
- Per visualizzare la pagina dei dettagli del cluster, fai clic sul nome di un cluster.
- Nella pagina dei dettagli del cluster, in Argomenti, fai clic sull'icona
BigQuery Export per qualsiasi argomento.
Si apre la pagina Crea un job Dataflow utilizzando il modello "Kafka to BigQuery".
Configura i campi nel modello in base alle informazioni incluse nelle sezioni seguenti.
Inserisci un nome job
Nel campo Nome job, inserisci un nome per il job Dataflow.
Il nome deve essere univoco tra tutti i job attualmente in esecuzione nel progetto.
Scegli un endpoint regionale per la pipeline
Nel campo Endpoint regionale, imposta l'endpoint regionale sulla località del cluster Kafka o del set di dati BigQuery per ridurre al minimo le tariffe di trasferimento dei dati tra regioni.
I worker Dataflow possono essere eseguiti indipendentemente dalla regione del cluster Kafka. Tuttavia, se avvii i worker al di fuori della regione del cluster Kafka, dovrai sostenere costi per il traffico in uscita tra regioni.
Per visualizzare la località del cluster, segui i passaggi descritti in Elencare i cluster Managed Service per Apache Kafka.
Scegli un modello Dataflow
Per modello Dataflow, mantieni il valore predefinito Kafka to BigQuery.
Configura origine
Per Origine, mantieni il valore predefinito Managed Service per Apache Kafka.
I valori predefiniti per Cluster Kafka, Argomento Kafka e Modalità di autenticazione dell'origine Kafka sono già selezionati in base all' argomento scelto. Mantieni questi valori.
Configura il formato dei messaggi Kafka
Il modello Dataflow supporta tre formati di messaggi:
Formato wire Avro Confluent: ogni messaggio Kafka include un byte magico, un ID schema e il record con codifica binaria Avro. Per impostazione predefinita, la chiave del messaggio non viene caricata in BigQuery.
Per i formati Avro (formato wire Confluent), puoi utilizzare un singolo schema o più schemi:
Singolo schema: tutti i messaggi rispettano un singolo schema Avro predefinito.
Più schemi: i messaggi possono utilizzare schemi diversi. Questa opzione è supportata solo per il formato wire Avro (Confluent).
Avro (con codifica binaria) : i messaggi contengono solo il payload del record senza metadati. Devi fornire un file dello schema Avro (.avsc) caricato in Cloud Storage. Tutti i messaggi devono rispettare questo singolo schema.
JSON: i record non richiedono uno schema predefinito. I record che non sono conformi allo schema della tabella BigQuery vengono inviati alla coda di messaggi non recapitabili (se configurata) oppure viene registrato un messaggio di errore. Il formato supportato è
{"field": "value"}. Il formato[{"name": "field", "value": "value"}]non è supportato.
Formato wire Avro Confluent
Se scegli questa opzione come Formato messaggi Kafka, configura le seguenti impostazioni aggiuntive:
Origine schema: questo campo indica alla pipeline dove trovare lo schema. Scegli una delle seguenti opzioni:
Registro di schema: gli schemi sono archiviati in un registro di schema Confluent. Questa opzione è utile per l'evoluzione degli schemi e la gestione di più versioni. Assicurati che il registro di schema sia accessibile alla rete del cluster Managed Service per Apache Kafka e che sia ospitato nella stessa regione dei worker Dataflow. Puoi utilizzare un registro di schema con scenari a schema singolo e multiplo. Configura le seguenti impostazioni aggiuntive:
URL di connessione al registro di schema: fornisci l'URL per connetterti al registro di schema.
Modalità di autenticazione: se il registro richiede l'autenticazione, seleziona OAuth o TLS. In caso contrario, seleziona Nessuna.
File di schema singolo: scegli questa opzione se tutti i messaggi seguono un singolo schema fisso definito in un file.
- File di Cloud Storage per il file dello schema Avro: il percorso del file dello schema Avro utilizzato per decodificare tutti i messaggi in un argomento.
Google Cloud Managed Service per Apache Kafka non offre un registro di schema. Il modello supporta solo il passaggio delle credenziali di autenticazione ai registri di schema compatibili con il formato wire Confluent.
Codifica binaria Avro
Se scegli questa opzione come Formato messaggi Kafka, configura le seguenti impostazioni aggiuntive:
- File di Cloud Storage per il file dello schema Avro: il percorso del file dello schema Avro utilizzato per decodificare tutti i messaggi in un argomento.
JSON
Se scegli questa opzione come Formato messaggi Kafka, non sono necessarie altre configurazioni.
Configura la persistenza della chiave Kafka nella tabella BigQuery
Se abiliti l'opzione Persisti la chiave del messaggio Kafka nella tabella BigQuery , devi anche aggiungere un campo denominato _key di tipo BYTES nella tabella di destinazione.
Sebbene la chiave possa rappresentare dati strutturati, il modello la considera come un array di byte.
Specifica le opzioni di offset
Per evitare di rielaborare i messaggi quando è necessario riavviare i singoli worker o l'intera pipeline, seleziona l'opzione Esegui il commit degli offset in Kafka. In questo modo, la pipeline riprende l'elaborazione dal punto in cui è stata interrotta, evitando l'elaborazione duplicata e potenziali incoerenze dei dati.
Nel campo Inserisci l'ID gruppo di consumer, inserisci un nome univoco per il gruppo di questa pipeline. Nella maggior parte dei casi, vuoi che la pipeline legga ogni messaggio una volta e che sia riavviabile.
Per il campo Offset di avvio Kafka predefinito, la pipeline Dataflow offre due opzioni di offset di avvio. Seleziona una delle seguenti opzioni:
Earliest: elabora i messaggi dall'inizio dell'argomento Kafka.
Latest: elabora i messaggi a partire dall'offset disponibile più recente.
Configura destinazione
Queste opzioni controllano il modo in cui la pipeline di dati scrive i dati in BigQuery.
Il campo Strategia per i nomi delle tabelle determina il modo in cui i dati vengono organizzati nelle tabelle BigQuery. Hai due opzioni:
Nome di una singola tabella (impostazione predefinita): tutti i dati dell'argomento Kafka vengono inseriti in una singola tabella BigQuery. Questa è l'opzione più semplice e consigliata quando i dati Kafka hanno lo stesso schema.
In Tabella di output BigQuery, specifica il nome della tabella in cui verranno archiviati tutti i dati dell'argomento Kafka.
Nomi delle tabelle dinamici: crea più tabelle BigQuery in base allo schema dei messaggi Kafka. Ti consigliamo di utilizzare questa opzione quando i messaggi Kafka hanno schemi diversi. Questa opzione funziona solo se la pipeline utilizza il formato wire Avro Confluent o il registro di schema per la gestione della struttura dei messaggi.
Ogni schema univoco nei messaggi Kafka ha una propria tabella in BigQuery. I nomi delle tabelle vengono generati automaticamente in base ai nomi degli schemi.
Se selezioni Nomi delle tabelle dinamici, configura questi campi aggiuntivi:
Progetto di output BigQuery: il Google Cloud progetto in cui si trova il set di dati BigQuery.
Set di dati di output BigQuery: il set di dati all'interno del progetto in cui vengono create le tabelle.
Prefisso per la denominazione delle tabelle BigQuery (facoltativo): aggiungi un prefisso ai nomi delle tabelle generati automaticamente per una migliore organizzazione.
Configura l'API BigQuery Storage Write
Per l'opzione Numero di stream per l'API BigQuery Storage Write, inizia con 0 (impostazione predefinita). Questa opzione consente alla pipeline di determinare automaticamente il numero ottimale di stream.
Questa impostazione controlla il numero di stream paralleli utilizzati per scrivere i dati in BigQuery. Numeri più elevati possono migliorare la velocità di scrittura, soprattutto per grandi volumi di dati. Il numero ottimale dipende dai dati e dalla configurazione di BigQuery.
Configura la frequenza di attivazione
Per l'opzione Frequenza di attivazione in secondi per l'API BigQuery Write , inizia con 5 secondi (impostazione predefinita). Questo è un buon compromesso per la maggior parte dei casi d'uso.
Questa impostazione determina la frequenza con cui i dati vengono scritti in BigQuery.
Numeri più bassi indicano scritture più frequenti. Questo può ridurre la latenza, ma potrebbe aumentare i costi di BigQuery. Numeri più elevati indicano scritture meno frequenti. Questa opzione può essere più conveniente, ma potrebbe aumentare la latenza.
Configura la coda di messaggi non recapitabili
A volte i messaggi non possono essere elaborati a causa di danneggiamenti, tipi di dati incompatibili o mancata corrispondenza dello schema con la tabella BigQuery di destinazione.
Per gestire questi casi, abilita la coda di messaggi non recapitabili nel modello e fornisci un nome della tabella. Il modello crea la tabella utilizzando uno schema standardizzato. I messaggi errati vengono scritti in una tabella BigQuery separata.
Configura la crittografia
Per impostazione predefinita, tutti i dati at-rest e in transito vengono criptati da a Google-owned and Google-managed encryption key. Se hai chiavi di crittografia gestite dal cliente (CMEK), puoi selezionare le tue chiavi. Per ulteriori informazioni su come configurare una CMEK, consulta Configurare la crittografia dei messaggi.
Configura il networking
Devi specificare la rete e la subnet del cluster nel modello Dataflow. La sezione Parametri facoltativi del modello consente di definire la rete per i worker Dataflow.
Per impostazione predefinita, il modello Dataflow da Kafka a BigQuery esegue il provisioning dei worker Dataflow nella rete predefinita del progetto. Per consentire al cluster Managed Service per Apache Kafka di inviare dati a BigQuery tramite Dataflow, assicurati che i worker Dataflow possano accedere alla rete del cluster.
Ti consigliamo di utilizzare la rete predefinita del progetto per il cluster Kafka se quest'ultimo non è connesso a una subnet nella rete predefinita del progetto.
Per ulteriori informazioni sulla configurazione del networking con la pipeline Dataflow, consulta:
Se riscontri difficoltà nella configurazione del networking Dataflow, consulta la guida alla risoluzione dei problemi di networking Dataflow.
Configura i parametri Dataflow facoltativi
Configura i parametri facoltativi solo se conosci l'impatto della configurazione sui worker Dataflow. Impostazioni errate possono influire su prestazioni o costi. Per spiegazioni dettagliate di ogni opzione, consulta Parametri facoltativi.
Monitoraggio
Il modello Dataflow da Kafka a BigQuery offre un'esperienza di monitoraggio che ti consente di esplorare log, metriche ed errori all'interno della console. Questa suite di strumenti di monitoraggio è disponibile come parte dell'interfaccia utente di Dataflow.
La scheda Metriche job ti consente di creare dashboard personalizzate. Per il modello da Kafka a BigQuery, ti consigliamo di configurare una dashboard Metriche job che monitori quanto segue:
Velocità effettiva: il volume di dati elaborati in un determinato momento. Questa metrica è utile per monitorare il flusso di dati attraverso il job e identificare potenziali problemi di prestazioni.
Per ulteriori informazioni, consulta Monitoraggio della velocità effettiva di Dataflow.
Aggiornamento dei dati: la differenza in secondi tra il timestamp dell' elemento dati e il momento in cui l'evento viene elaborato nella pipeline. Questa metrica aiuta a identificare i colli di bottiglia delle prestazioni e delle origini dati o i nuovi tentativi frequenti.
Per ulteriori informazioni, consulta Monitoraggio dell'aggiornamento dei dati di Dataflow.
Backlog: la quantità di byte in attesa di elaborazione. Queste informazioni vengono utilizzate per le decisioni di scalabilità automatica.
Per ulteriori informazioni sul monitoraggio di Dataflow, consulta la documentazione relativa al monitoraggio di Dataflow.
Risoluzione dei problemi
Se riscontri problemi di prestazioni con la pipeline Dataflow, Dataflow fornisce un set completo di strumenti di risoluzione dei problemi e diagnostici.
Di seguito sono riportati due scenari comuni e le rispettive guide alla risoluzione dei problemi:
Risolvere i problemi relativi alle pipeline di flussi di dati lente o bloccate.
Risolvere i problemi relativi ai ritardatari nelle pipeline.
Per una panoramica generale del debug delle pipeline Dataflow, consulta Risolvere i problemi ed eseguire il debug delle pipeline Dataflow.
Limitazioni note
Il modello non supporta il passaggio delle credenziali per l'autenticazione al registro di schema.
Quando crei il job Dataflow da Kafka a BigQuery, assicurati che il Google Cloud progetto sia impostato sullo stesso progetto che contiene il cluster Managed Service per Apache Kafka.