Se devi caricare messaggi AVRO o JSON da un argomento di 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 console Google Cloud . Puoi anche utilizzare l'API REST o Google Cloud CLI per configurare questa pipeline.
Google Cloud prodotti utilizzati
Il modello Kafka to BigQuery Dataflow utilizza i seguenti prodotti Google Cloud fatturabili. Utilizza il Calcolatore prezzi per generare una stima dei costi in base all'utilizzo previsto.
- BigQuery: BigQuery è il data warehouse serverless di 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. Kafka to BigQuery Dataflow 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 auto-ripristino di Dataflow garantiscono l'esecuzione affidabile ed efficiente della pipeline.
- Cloud Storage: sebbene non sia direttamente coinvolto nel flusso di dati principale, Cloud Storage può essere utilizzato per archiviare i file di schema Avro se utilizzi la codifica binaria Avro per i messaggi Kafka. In questo modo, la pipeline Dataflow può interpretare correttamente la struttura dei dati.
Inoltre, la soluzione utilizza anche Managed Service per Apache Kafka.
- 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 dati di streaming, anziché sulla gestione dell'infrastruttura. Per ulteriori informazioni sui prezzi di Managed Service per Apache Kafka, consulta la guida ai prezzi.
Prima di iniziare
Prima di avviare il modello Kafka to BigQuery Dataflow, 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 Configurare il formato dei messaggi Kafka.
Abilita le seguenti API Google Cloud :
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 a quello 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 saperne di più su come creare un set di dati e una tabella BigQuery, consulta Creare un set di dati e Creare una tabella vuota con una definizione dello schema.
Concedi il ruolo Managed Kafka Client 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 Managed Kafka Client al account di servizio Dataflow, segui questi passaggi:
Console
- Nella console Google Cloud vai alla pagina IAM.
Vai a IAM - Verifica che il progetto sia impostato sul progetto consumer a cui accederà il client Managed Service per Apache Kafka.
- 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 console Google Cloud , attiva Cloud Shell.
Nella parte inferiore della console Google Cloud 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à installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.
-
Esegui il comando
gcloud projects add-iam-policy-binding: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 serviziot worker Dataflow.
-
Avvia il modello Dataflow Kafka to BigQuery
Puoi avviare il modello Kafka to BigQuery Dataflow dalla pagina dei dettagli del cluster nella console.
-
Nella console Google Cloud , vai alla pagina Cluster.
Vengono elencati i cluster che hai creato in un progetto.
- Per visualizzare la pagina dei dettagli del cluster, fai clic sul nome di un cluster.
- Nella pagina dei dettagli del cluster, per Argomenti, fai clic sull'icona di
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 per il 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.
Scegliere un endpoint regionale per la pipeline
Per il campo Endpoint regionale, imposta l'endpoint regionale sulla posizione del cluster Kafka o del set di dati BigQuery per ridurre al minimo le tariffe per il trasferimento di dati tra regioni.
I worker Dataflow possono essere eseguiti indipendentemente dalla regione del cluster Kafka. Tuttavia, se avvii worker al di fuori della regione del cluster Kafka, devi sostenere i costi del traffico in uscita tra regioni.
Per visualizzare la posizione del cluster, segui i passaggi descritti in Elenca i cluster Managed Service per Apache Kafka.
Scegliere un modello Dataflow
In Modello Dataflow, mantieni il valore predefinito Kafka to BigQuery.
Configura origine
Per Origine, mantieni il valore predefinito di 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 che hai scelto. Conserva questi valori.
Configurare il formato dei messaggi Kafka
Il modello Dataflow supporta tre formati di messaggio:
Formato di trasmissione 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 di trasferimento Confluent), puoi utilizzare uno schema singolo o più schemi:
Schema singolo: tutti i messaggi rispettano un unico schema Avro predefinito.
Più schemi: i messaggi possono utilizzare schemi diversi. Questa funzionalità è supportata solo per Avro (formato wire Confluent).
Avro (codifica binaria): i messaggi contengono solo il payload del record senza metadati. Devi fornire un file di 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 rispettano lo schema della tabella BigQuery vengono inviati alla coda dei messaggi non recapitabili (se configurata) o viene registrato un messaggio di errore. Il formato supportato è
{"field": "value"}. Il formato[{"name": "field", "value": "value"}]non è supportato.
Formato di trasferimento 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. Ciò è 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 degli schemi 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 tuo registro di schema.
Modalità di autenticazione: se il registro richiede l'autenticazione, seleziona OAuth o TLS. In caso contrario, seleziona Nessuno.
File di schema singolo: scegli questa opzione se tutti i tuoi messaggi seguono uno schema fisso e singolo definito in un file.
- File di Cloud Storage nel file dello schema Avro: il percorso del file dello schema Avro utilizzato per decodificare tutti i messaggi in un argomento.
Managed Service per Apache Kafka non offre un registro di schema. Il modello supporta solo il trasferimento delle credenziali di autenticazione ai registri di schema compatibili con il formato Confluent-wire.
Codifica binaria Avro
Se scegli questa opzione come formato messaggi Kafka, configura le seguenti impostazioni aggiuntive:
- File di Cloud Storage nel 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 dei messaggi Kafka, non sono necessarie altre configurazioni.
Configura la persistenza della chiave Kafka nella tabella BigQuery
Se abiliti l'opzione
Mantieni 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 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 l'aveva interrotta, evitando l'elaborazione duplicata e potenziali incongruenze dei dati.
Nel campo Inserisci ID gruppo di consumer, inserisci un nome univoco per il gruppo della pipeline. Nella maggior parte dei casi, vuoi che la pipeline legga ogni messaggio una volta e che sia riavviabile.
Per il campo Offset iniziale Kafka predefinito, la pipeline Dataflow offre due opzioni di offset iniziale. Seleziona una delle seguenti opzioni:
Earliest: elabora i messaggi dall'inizio dell'argomento Kafka.
Più recente: elabora i messaggi a partire dall'offset più recente disponibile.
Configura destinazione
Queste opzioni controllano il modo in cui la pipeline di dati scrive i dati in BigQuery.
Il campo Strategia per il nome della tabella determina la modalità di organizzazione dei dati 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.
Per 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 questa opzione quando i messaggi Kafka hanno schemi diversi. Questa opzione funziona solo se la pipeline utilizza Avro Confluent Wire o Schema Registry 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 progetto Google Cloud 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 Numero di stream per l'API BigQuery Storage Write, inizia con 0 (valore predefinito). Questa opzione consente alla pipeline di determinare automaticamente il numero ottimale di stream.
Questa impostazione controlla quanti flussi paralleli vengono utilizzati per scrivere dati in BigQuery. Numeri più elevati possono migliorare la velocità di scrittura, soprattutto per grandi volumi di dati. Il numero ottimale dipende dai tuoi dati e dalla configurazione di BigQuery.
Configurare la frequenza di attivazione
Per l'opzione Frequenza di attivazione in secondi per l'API BigQuery Write, inizia con 5 secondi (valore predefinito). Si tratta di un buon equilibrio 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. Ciò può ridurre la latenza, ma potrebbe aumentare i costi di BigQuery. Numeri più alti indicano scritture meno frequenti. Questa operazione può essere più conveniente, ma potrebbe aumentare la latenza.
Configura coda messaggi non recapitabili
A volte i messaggi non possono essere elaborati a causa di danneggiamento, tipi di dati incompatibili o mancata corrispondenza dello schema con la tabella BigQuery di destinazione.
Per gestire questi casi, attiva la coda dei messaggi non recapitabili nel modello e fornisci un nome 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 a riposo e in transito sono criptati da un Google-owned and Google-managed encryption key. Se hai chiavi di crittografia gestite dal cliente (CMEK), puoi selezionare le tue chiavi. Per saperne di più su come configurare una chiave 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.
Il modello Dataflow da Kafka a BigQuery esegue il provisioning dei worker Dataflow nella rete predefinita del progetto per impostazione predefinita. 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.
Se il cluster Kafka non è connesso a una subnet nella rete predefinita del progetto, ti consigliamo di utilizzare la rete predefinita del progetto per il cluster Kafka.
Per saperne di più sulla configurazione del networking con la pipeline Dataflow, consulta quanto segue:
Se riscontri difficoltà nella configurazione della rete Dataflow, consulta la guida alla risoluzione dei problemi di rete Dataflow.
Configura i parametri Dataflow facoltativi
Configura i parametri facoltativi solo se conosci l'impatto della configurazione sui worker Dataflow. Impostazioni errate possono influire sulle prestazioni o sui costi. Per spiegazioni dettagliate di ogni opzione, vedi Parametri facoltativi.
Monitoraggio
Il modello Dataflow per 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 nell'interfaccia utente di Dataflow.
La scheda Metriche job consente di creare dashboard personalizzate. Per il modello da Kafka a BigQuery, ti consigliamo di configurare un dashboard delle metriche dei job che monitori quanto segue:
Throughput: il volume di dati elaborati in un determinato momento. Ciò è utile per monitorare il flusso di dati nel job e identificare potenziali problemi di prestazioni.
Per saperne di più, consulta Monitoraggio della velocità effettiva di Dataflow.
Aggiornamento dei dati: la differenza in secondi tra il timestamp dell'elemento di dati e l'ora in cui l'evento viene elaborato nella pipeline. Ciò è utile per identificare i colli di bottiglia delle prestazioni e delle origini dati o i tentativi frequenti.
Per saperne di più, consulta Monitoraggio dell'aggiornamento dei dati di Dataflow.
Backlog: la quantità di byte in attesa di elaborazione. Queste informazioni vengono utilizzate per prendere decisioni relative alla scalabilità automatica.
Per saperne di più sul monitoraggio di Dataflow, consulta la documentazione sul monitoraggio di Dataflow.
Risoluzione dei problemi
Se riscontri problemi di prestazioni con la tua pipeline Dataflow, Dataflow fornisce un set completo di strumenti di risoluzione dei problemi e diagnostica.
Di seguito sono riportati due scenari comuni e le rispettive guide alla risoluzione dei problemi:
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 trasferimento delle credenziali per l'autenticazione al tuo Schema Registry.
Quando crei il job Dataflow da Kafka a BigQuery, assicurati che il progetto Google Cloud sia impostato sullo stesso progetto che contiene il cluster Managed Service per Apache Kafka.