Scrivere i dati da Kafka a BigQuery utilizzando Dataflow

Questa pagina mostra come utilizzare Dataflow per leggere i dati dal servizio gestito Google Cloud per Apache Kafka e scrivere i record in una tabella BigQuery. Questo tutorial utilizza il modello Apache Kafka to BigQuery per creare il job Dataflow.

Panoramica

Apache Kafka è una piattaforma open source per lo streaming di eventi. Kafka viene comunemente utilizzato nelle architetture distribuite per consentire la comunicazione tra componenti a basso accoppiamento. Puoi utilizzare Dataflow per leggere gli eventi da Kafka, elaborarli e scrivere i risultati in una tabella BigQuery per ulteriori analisi.

Managed Service per Apache Kafka è un servizio che ti aiuta a eseguire cluster Kafka sicuri e scalabili. Google Cloud

Autorizzazioni obbligatorie

Il service account worker Dataflow deve avere i seguenti ruoli Identity and Access Management (IAM):

• Managed Kafka Client (roles/managedkafka.client)
• Editor dati BigQuery (roles/bigquery.dataEditor)

Per ulteriori informazioni, vedi Sicurezza e autorizzazioni di Dataflow.

Crea un cluster Kafka

In questo passaggio, crei un cluster Managed Service per Apache Kafka. Per maggiori informazioni, consulta Creare un cluster Managed Service per Apache Kafka.

gcloud managed-kafka clusters create CLUSTER \
--location=REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME

La creazione di un cluster richiede in genere 20-30 minuti.

Crea un argomento Kafka

Dopo aver creato il cluster Managed Service per Apache Kafka, crea un argomento.

gcloud managed-kafka topics create TOPIC_NAME \
--cluster=CLUSTER \
--location=REGION \
--partitions=10 \
--replication-factor=3

Crea una tabella BigQuery

In questo passaggio, crei una tabella BigQuery con lo schema seguente:

Se non hai ancora un set di dati BigQuery, creane uno. Per saperne di più, consulta Creare set di dati. Poi crea una nuova tabella vuota:

bq mk --table \
  PROJECT_ID:DATASET_NAME.TABLE_NAME \
  name:STRING,customer_id:INTEGER

Esegui il job Dataflow

Dopo aver creato il cluster Kafka e la tabella BigQuery, esegui il modello Dataflow.

gcloud dataflow flex-template run kafka-to-bq \
--template-file-gcs-location gs://dataflow-templates/latest/flex/Kafka_to_BigQuery \
--region LOCATION \
--parameters \
readBootstrapServerAndTopic=projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC,\
persistKafkaKey=false,\
writeMode=SINGLE_TABLE_NAME,\
kafkaReadOffset=earliest,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
outputTableSpec=PROJECT_ID:DATASET_NAME.TABLE_NAME\
useBigQueryDLQ=true,\
outputDeadletterTable=PROJECT_ID:DATASET_NAME.ERROR_TABLE_NAME

Inviare messaggi a Kafka

Dopo l'avvio del job Dataflow, puoi inviare messaggi a Kafka e la pipeline li scrive in BigQuery.

1. Crea una VM nella stessa subnet del cluster Kafka e installa gli strumenti a riga di comando Kafka. Per istruzioni dettagliate, vedi Configurare una macchina client in Pubblicare e utilizzare i messaggi con la CLI.

2. Esegui il comando seguente per scrivere messaggi nell'argomento Kafka:

   kafka-console-producer.sh \
    --topic TOPIC \
    --bootstrap-server bootstrap.CLUSTER_ID.LOCATION.managedkafka.PROJECT_ID.cloud.goog:9092 \
    --producer.config client.properties

   Sostituisci le seguenti variabili:

   • TOPIC: il nome dell'argomento Kafka
   • CLUSTER_ID: il nome del cluster
   • LOCATION: la regione in cui si trova il cluster
   • PROJECT_ID: il nome del tuo Google Cloud progetto

3. Al prompt, inserisci le seguenti righe di testo per inviare messaggi a Kafka:

   {"name": "Alice", "customer_id": 1}
   {"name": "Bob", "customer_id": 2}
   {"name": "Charles", "customer_id": 3}
   

Utilizzare una coda di messaggi non recapitabili

Durante l'esecuzione del job, la pipeline potrebbe non riuscire a scrivere singoli messaggi in BigQuery. I possibili errori includono:

• Errori di serializzazione, incluso JSON con formattazione errata.
• Errori di conversione del tipo, causati da una mancata corrispondenza tra lo schema della tabella e i dati JSON.
• Campi aggiuntivi nei dati JSON che non sono presenti nello schema della tabella.

Questi errori non causano l'interruzione del job e non vengono visualizzati come errori nel log del job Dataflow. La pipeline utilizza invece una coda di messaggi non recapitabili per gestire questi tipi di errori.

Per abilitare la coda dei messaggi non recapitabili quando esegui il modello, imposta i seguenti parametri del modello:

• useBigQueryDLQ: true
• outputDeadletterTable: un nome di tabella BigQuery completo; ad esempio, my-project:dataset1.errors

La pipeline crea automaticamente la tabella. Se si verifica un errore durante l'elaborazione di un messaggio Kafka, la pipeline scrive una voce di errore nella tabella.

Esempi di messaggi di errore:

Utilizzare i tipi di dati BigQuery

Internamente, il connettore I/O Kafka converte i payload dei messaggi JSON in oggetti Apache Beam TableRow e traduce i valori dei campi TableRow in tipi BigQuery.

La tabella seguente mostra le rappresentazioni JSON dei tipi di dati di BigQuery.

Dati strutturati

Se i tuoi messaggi JSON seguono uno schema coerente, puoi rappresentare gli oggetti JSON utilizzando il tipo di dati STRUCT in BigQuery.

Nel seguente esempio, il campo answers è un oggetto JSON con due sottocampi, a e b:

{"name":"Emily","answers":{"a":"yes","b":"no"}}

La seguente istruzione SQL crea una tabella BigQuery con uno schema compatibile:

CREATE TABLE my_dataset.kafka_events (name STRING, answers STRUCT<a STRING, b STRING>);

La tabella risultante sarà simile alla seguente:

+-------+----------------------+
| name  |       answers        |
+-------+----------------------+
| Emily | {"a":"yes","b":"no"} |
+-------+----------------------+

e semistrutturati

Se i tuoi messaggi JSON non seguono uno schema rigoroso, valuta la possibilità di archiviarli in BigQuery come tipo di dati JSON. Se memorizzi i dati JSON come tipo JSON, non devi definire lo schema in anticipo. Dopo l'importazione dati, puoi eseguire query sui dati utilizzando gli operatori di accesso ai campi (notazione con punto) e di accesso agli array in GoogleSQL. Per ulteriori informazioni, consulta Utilizzo dei dati JSON in GoogleSQL.

Utilizzare una UDF per trasformare i dati

Questo tutorial presuppone che i messaggi Kafka siano formattati come JSON e che lo schema della tabella BigQuery corrisponda ai dati JSON, senza che siano state applicate trasformazioni ai dati.

Se vuoi, puoi fornire una funzione definita dall'utente;utente (UDF) JavaScript che trasforma i dati prima che vengano scritti in BigQuery. La funzione definita dall'utente può anche eseguire un'ulteriore elaborazione, ad esempio il filtraggio, la rimozione delle informazioni che consentono l'identificazione personale (PII) o l'arricchimento dei dati con campi aggiuntivi.

Per ulteriori informazioni, consulta Creare funzioni definite dall'utente per i modelli Dataflow.

Passaggi successivi

• Utilizzare Dataflow con Managed Service per Apache Kafka
• Scopri di più su BigQuery